Categories

Archives

A sample text widget

Etiam pulvinar consectetur dolor sed malesuada. Ut convallis euismod dolor nec pretium. Nunc ut tristique massa.

Nam sodales mi vitae dolor ullamcorper et vulputate enim accumsan. Morbi orci magna, tincidunt vitae molestie nec, molestie at mi. Nulla nulla lorem, suscipit in posuere in, interdum non magna.

Building AI Agents with Tool Use: From ReAct to Production Systems

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.

AI Agents with Tool Use Architecture
AI Agents: Reasoning, Planning, and Tool Execution

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

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

You can use these HTML tags

<a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <s> <strike> <strong>

  

  

  

This site uses Akismet to reduce spam. Learn how your comment data is processed.