Introduction: AI agents represent the next evolution beyond simple chatbots—they can reason about problems, break them into steps, use external tools, and iterate until they achieve a goal. Unlike traditional LLM applications that respond to a single prompt, agents maintain state, make decisions, and take actions in the real world. The key innovation is tool use: giving LLMs the ability to search the web, execute code, query databases, and interact with APIs. This guide covers the fundamentals of building AI agents, from the ReAct pattern to production-ready implementations with LangChain and OpenAI’s function calling.

The ReAct Pattern
ReAct (Reasoning + Acting) is the foundational pattern for AI agents. The agent follows a loop: think about what to do next (Thought), take an action using a tool (Action), observe the result (Observation), and repeat until the task is complete. This explicit reasoning trace makes agents more interpretable and allows them to recover from errors.
from openai import OpenAI
import json
client = OpenAI()
# Define available tools
tools = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "Search the web for current information",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "execute_python",
"description": "Execute Python code and return the result",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to execute"}
},
"required": ["code"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read contents of a file",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "File path to read"}
},
"required": ["path"]
}
}
}
]
# Tool implementations
def search_web(query: str) -> str:
# In production, use a real search API
return f"Search results for '{query}': [Result 1, Result 2, Result 3]"
def execute_python(code: str) -> str:
try:
# WARNING: In production, use a sandboxed environment
local_vars = {}
exec(code, {"__builtins__": {}}, local_vars)
return str(local_vars.get("result", "Code executed successfully"))
except Exception as e:
return f"Error: {str(e)}"
def read_file(path: str) -> str:
try:
with open(path, 'r') as f:
return f.read()[:1000] # Limit output
except Exception as e:
return f"Error reading file: {str(e)}"
tool_functions = {
"search_web": search_web,
"execute_python": execute_python,
"read_file": read_file
}
def run_agent(user_query: str, max_iterations: int = 10) -> str:
"""Run an agent loop until task completion."""
messages = [
{"role": "system", "content": """You are a helpful AI agent that can use tools to accomplish tasks.
Think step by step about what you need to do.
Use tools when needed to gather information or take actions.
When you have enough information to answer, provide your final response."""},
{"role": "user", "content": user_query}
]
for i in range(max_iterations):
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# Check if agent wants to use tools
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"[Agent] Calling {function_name} with {arguments}")
# Execute the tool
result = tool_functions[function_name](**arguments)
# Add tool result to messages
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
else:
# No tool calls - agent is done
return message.content
return "Max iterations reached"
# Example usage
result = run_agent("What's the current weather in Tokyo and convert 25°C to Fahrenheit?")
print(result)
Building Agents with LangChain
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool, StructuredTool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools import DuckDuckGoSearchRun
from pydantic import BaseModel, Field
import subprocess
# Initialize LLM
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
# Define tools
search = DuckDuckGoSearchRun()
class CalculatorInput(BaseModel):
expression: str = Field(description="Mathematical expression to evaluate")
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
try:
# Only allow safe operations
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression)
return str(result)
except Exception as e:
return f"Error: {str(e)}"
class ShellInput(BaseModel):
command: str = Field(description="Shell command to execute")
def run_shell(command: str) -> str:
"""Execute a shell command (use with caution)."""
# Whitelist safe commands
safe_commands = ["ls", "pwd", "date", "whoami", "cat", "head", "tail", "wc"]
cmd_parts = command.split()
if not cmd_parts or cmd_parts[0] not in safe_commands:
return f"Error: Command '{cmd_parts[0]}' not in allowed list"
try:
result = subprocess.run(
command, shell=True, capture_output=True, text=True, timeout=10
)
return result.stdout or result.stderr
except Exception as e:
return f"Error: {str(e)}"
tools = [
Tool(
name="search",
func=search.run,
description="Search the web for current information. Input should be a search query."
),
StructuredTool.from_function(
func=calculate,
name="calculator",
description="Evaluate mathematical expressions. Input should be a valid math expression.",
args_schema=CalculatorInput
),
StructuredTool.from_function(
func=run_shell,
name="shell",
description="Execute safe shell commands (ls, pwd, date, cat, etc.)",
args_schema=ShellInput
)
]
# Create agent prompt
prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful AI assistant with access to tools.
Use tools when you need to search for information, perform calculations, or interact with the system.
Always explain your reasoning before taking actions.
If you're unsure, search for more information before answering."""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# Create and run agent
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True
)
# Run the agent
result = agent_executor.invoke({
"input": "What is 15% of 847.50? Also, what files are in the current directory?"
})
print(result["output"])
Custom Tool Creation
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type
import requests
import sqlite3
# Custom tool with complex logic
class DatabaseQueryInput(BaseModel):
query: str = Field(description="SQL SELECT query to execute")
database: str = Field(default="main.db", description="Database file path")
class DatabaseQueryTool(BaseTool):
name: str = "database_query"
description: str = "Execute read-only SQL queries against a SQLite database"
args_schema: Type[BaseModel] = DatabaseQueryInput
def _run(self, query: str, database: str = "main.db") -> str:
# Security: Only allow SELECT queries
if not query.strip().upper().startswith("SELECT"):
return "Error: Only SELECT queries are allowed"
try:
conn = sqlite3.connect(database)
cursor = conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
# Format as table
result = " | ".join(columns) + "\n"
result += "-" * len(result) + "\n"
for row in rows:
result += " | ".join(str(v) for v in row) + "\n"
conn.close()
return result
except Exception as e:
return f"Database error: {str(e)}"
# API integration tool
class WeatherInput(BaseModel):
city: str = Field(description="City name for weather lookup")
class WeatherTool(BaseTool):
name: str = "get_weather"
description: str = "Get current weather for a city"
args_schema: Type[BaseModel] = WeatherInput
api_key: str = ""
def _run(self, city: str) -> str:
try:
# Using OpenWeatherMap API
url = f"http://api.openweathermap.org/data/2.5/weather"
params = {"q": city, "appid": self.api_key, "units": "metric"}
response = requests.get(url, params=params, timeout=10)
data = response.json()
if response.status_code == 200:
return f"""Weather in {city}:
Temperature: {data['main']['temp']}°C
Feels like: {data['main']['feels_like']}°C
Humidity: {data['main']['humidity']}%
Conditions: {data['weather'][0]['description']}"""
else:
return f"Error: {data.get('message', 'Unknown error')}"
except Exception as e:
return f"Error fetching weather: {str(e)}"
# File manipulation tool with safety checks
class FileWriteInput(BaseModel):
path: str = Field(description="File path to write to")
content: str = Field(description="Content to write")
mode: str = Field(default="w", description="Write mode: 'w' for overwrite, 'a' for append")
class SafeFileWriteTool(BaseTool):
name: str = "write_file"
description: str = "Write content to a file (restricted to safe directories)"
args_schema: Type[BaseModel] = FileWriteInput
allowed_dirs: list = ["/tmp", "./output"]
def _run(self, path: str, content: str, mode: str = "w") -> str:
import os
# Security checks
abs_path = os.path.abspath(path)
if not any(abs_path.startswith(os.path.abspath(d)) for d in self.allowed_dirs):
return f"Error: Path must be in allowed directories: {self.allowed_dirs}"
if mode not in ["w", "a"]:
return "Error: Mode must be 'w' or 'a'"
try:
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
with open(abs_path, mode) as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to {path}"
except Exception as e:
return f"Error writing file: {str(e)}"
# Use custom tools
tools = [
DatabaseQueryTool(),
WeatherTool(api_key="your-api-key"),
SafeFileWriteTool()
]
Agent Memory and Context
from langchain.memory import ConversationBufferWindowMemory, ConversationSummaryMemory
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
# Window memory - keeps last N interactions
window_memory = ConversationBufferWindowMemory(
memory_key="chat_history",
return_messages=True,
k=10 # Keep last 10 exchanges
)
# Summary memory - summarizes older conversations
summary_memory = ConversationSummaryMemory(
llm=llm,
memory_key="chat_history",
return_messages=True
)
# Agent with memory
prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful AI assistant with memory of our conversation.
Use the chat history to maintain context and provide consistent responses.
Reference previous discussions when relevant."""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=window_memory,
verbose=True
)
# Multi-turn conversation
agent_executor.invoke({"input": "My name is Alex and I'm working on a Python project"})
agent_executor.invoke({"input": "What's my name and what am I working on?"})
agent_executor.invoke({"input": "Can you help me with error handling in my project?"})
Error Handling and Reliability
from langchain.agents import AgentExecutor
from langchain.callbacks import get_openai_callback
import time
def run_agent_with_retry(
agent_executor: AgentExecutor,
input_text: str,
max_retries: int = 3,
timeout_seconds: int = 60
) -> dict:
"""Run agent with retry logic and timeout."""
last_error = None
for attempt in range(max_retries):
try:
with get_openai_callback() as cb:
start_time = time.time()
result = agent_executor.invoke(
{"input": input_text},
config={"max_execution_time": timeout_seconds}
)
elapsed = time.time() - start_time
return {
"success": True,
"output": result["output"],
"tokens_used": cb.total_tokens,
"cost": cb.total_cost,
"elapsed_seconds": elapsed,
"attempts": attempt + 1
}
except Exception as e:
last_error = e
print(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt < max_retries - 1:
wait_time = 2 ** attempt # Exponential backoff
print(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
return {
"success": False,
"error": str(last_error),
"attempts": max_retries
}
# Graceful degradation
class RobustAgent:
def __init__(self, agent_executor: AgentExecutor, fallback_llm):
self.agent = agent_executor
self.fallback = fallback_llm
def run(self, query: str) -> str:
# Try agent first
result = run_agent_with_retry(self.agent, query)
if result["success"]:
return result["output"]
# Fall back to simple LLM response
print("Agent failed, falling back to simple LLM...")
response = self.fallback.invoke(query)
return f"[Fallback response] {response.content}"
References
- ReAct Paper: https://arxiv.org/abs/2210.03629
- LangChain Agents: https://python.langchain.com/docs/modules/agents/
- OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
- Tool Use Best Practices: https://docs.anthropic.com/claude/docs/tool-use
Conclusion
AI agents with tool use represent a fundamental shift from passive question-answering to active problem-solving. By combining LLM reasoning with the ability to search, compute, and interact with external systems, agents can tackle complex tasks that require multiple steps and real-world information. Start with simple tools like search and calculation, then gradually add more capabilities as you understand the patterns. Remember that reliability is crucial—implement proper error handling, timeouts, and fallbacks. The ReAct pattern provides a solid foundation, while frameworks like LangChain accelerate development. As you build more sophisticated agents, focus on safety: validate inputs, restrict tool capabilities, and always maintain human oversight for critical operations.
Discover more from Code, Cloud & Context
Subscribe to get the latest posts sent to your email.

Leave a Reply