Building AI Agents

An AI agent is an LLM-powered system that can reason about a task, decide which actions to take, execute those actions using tools, and observe the results to inform its next step. Unlike simple chains with fixed steps, agents dynamically choose their path based on the situation.

The ReAct Pattern

ReAct (Reasoning + Acting) is the foundational pattern for LLM agents. The LLM alternates between:

1. Thought --- reasoning about what to do next 2. Action --- choosing and executing a tool 3. Observation --- processing the tool's output 4. Repeat until the task is complete

User: What's the weather in Paris and what should I pack?
Thought: I need to check the weather in Paris first.
Action: get_weather(city="Paris")
Observation: Paris: 15°C, partly cloudy, 30% chance of rain
Thought: It's cool with possible rain. I should suggest appropriate clothing.
Action: (no tool needed, I can reason directly)
Answer: Paris is 15°C and partly cloudy with a 30% chance of rain.
        Pack layers, a light jacket, and an umbrella.

Tool Use / Function Calling

Modern LLMs support native function calling, where the model outputs structured tool calls instead of free text.

from openai import OpenAI
import json
client = OpenAI()
Define tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "The city name",
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature units",
                    },
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query",
                    },
                },
                "required": ["query"],
            },
        },
    },
]
Tool implementations
def get_weather(city: str, units: str = "celsius") -> str:
    # In production, call a weather API
    return json.dumps({"city": city, "temp": 15, "condition": "partly cloudy", "units": units})
def search_web(query: str) -> str:
    return json.dumps({"results": [f"Result for: {query}"]})
tool_map = {"get_weather": get_weather, "search_web": search_web}
Agent loop
messages = [{"role": "user", "content": "What's the weather in Paris?"}]
while True:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
    )
    message = response.choices[0].message
    messages.append(message)
    # If no tool calls, we're done
    if not message.tool_calls:
        print(f"Agent: {message.content}")
        break
    # Execute each tool call
    for tool_call in message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        print(f"Calling: {func_name}({func_args})")
        result = tool_map[func_name](**func_args)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result,
        })

Building Agents with LangChain

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
llm = ChatOpenAI(model="gpt-4o-mini")
Define tools using the @tool decorator
@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression. Input should be a valid Python math expression."""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"
@tool
def get_word_count(text: str) -> str:
    """Count the number of words in a text string."""
    return str(len(text.split()))
@tool
def reverse_string(text: str) -> str:
    """Reverse a string."""
    return text[::-1]
tools = [calculator, get_word_count, reverse_string]
Create the agent
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools. Use them when needed."),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,     # See reasoning steps
    max_iterations=5, # Prevent infinite loops
)
Run the agent
result = agent_executor.invoke({
    "input": "What is 2^10 + 3^5, and how many words are in the sentence 'The quick brown fox jumps over the lazy dog'?"
})
print(result["output"])

LangGraph: Stateful Agent Workflows

LangGraph extends LangChain with graph-based workflows, giving you explicit control over agent state and transitions.

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
import operator
Define state schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
Define tools
@tool
def search(query: str) -> str:
    """Search for information on the web."""
    return f"Search results for '{query}': RAG is a technique that combines retrieval with generation."
@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))
tools = [search, calculate]
tool_node = ToolNode(tools)
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
Define the agent node
def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
Define routing logic
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END
Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
Compile and run
app = workflow.compile()
from langchain_core.messages import HumanMessage
result = app.invoke({"messages": [HumanMessage(content="Search for what RAG is, then calculate 2+2")]})
for msg in result["messages"]:
    print(f"{msg.type}: {msg.content[:100] if msg.content else '[tool call]'}")

Multi-Agent Systems

Multi-agent architectures use multiple specialized agents that collaborate to solve complex tasks.

CrewAI

from crewai import Agent, Task, Crew, Process
Define specialized agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find and analyze the latest AI trends and developments",
    backstory="You are an expert research analyst with deep knowledge of AI.",
    verbose=True,
    allow_delegation=False,
)
writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging technical content from research findings",
    backstory="You are a skilled technical writer who makes complex topics accessible.",
    verbose=True,
    allow_delegation=False,
)
reviewer = Agent(
    role="Editor",
    goal="Review and improve content for accuracy and clarity",
    backstory="You are a meticulous editor with expertise in technical content.",
    verbose=True,
    allow_delegation=False,
)
Define tasks
research_task = Task(
    description="Research the current state of AI agents in 2024-2025. Focus on key frameworks, patterns, and real-world applications.",
    expected_output="A detailed summary of AI agent trends with specific examples.",
    agent=researcher,
)
writing_task = Task(
    description="Write a blog post about AI agents based on the research findings. Make it engaging and informative.",
    expected_output="A well-structured blog post of 500-800 words.",
    agent=writer,
    context=[research_task],  # Uses output from research_task
)
review_task = Task(
    description="Review the blog post for technical accuracy, clarity, and engagement. Provide the final polished version.",
    expected_output="The final reviewed and polished blog post.",
    agent=reviewer,
    context=[writing_task],
)
Create the crew
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # Tasks run in order
    verbose=True,
)
Execute
result = crew.kickoff()
print(result)

Planning Strategies

Agents need planning strategies for complex multi-step tasks:

Plan-and-Execute

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
llm = ChatOpenAI(model="gpt-4o", temperature=0)
class Plan(BaseModel):
    steps: list[str] = Field(description="Ordered list of steps to accomplish the task")
    reasoning: str = Field(description="Why this plan will work")
planner_prompt = ChatPromptTemplate.from_template(
    """You are a planning agent. Given a task, create a step-by-step plan.
    Each step should be a single action that can be executed.
    Task: {task}
    Available tools: {tools}
    Create a plan:"""
)
planner = planner_prompt | llm.with_structured_output(Plan)
plan = planner.invoke({
    "task": "Research the latest LLM benchmarks and create a comparison table",
    "tools": "search_web, read_page, create_table, write_file",
})
print("Plan:")
for i, step in enumerate(plan.steps, 1):
    print(f"  {i}. {step}")
print(f"\nReasoning: {plan.reasoning}")

Guardrails

Guardrails prevent agents from taking harmful or unintended actions:

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from pydantic import BaseModel, Field
Input validation guardrail
class SafeSearchInput(BaseModel):
    query: str = Field(
        description="Search query",
        max_length=500,
    )
    def validate_query(self):
        blocked_terms = ["hack", "exploit", "bypass security"]
        for term in blocked_terms:
            if term.lower() in self.query.lower():
                raise ValueError(f"Query contains blocked term: {term}")
        return True
@tool(args_schema=SafeSearchInput)
def safe_search(query: str) -> str:
    """Search the web safely with input validation."""
    validated = SafeSearchInput(query=query)
    validated.validate_query()
    return f"Results for: {query}"
Output guardrail
def check_output_safety(response: str) -> str:
    """Check if agent output contains sensitive information."""
    import re
    # Check for PII patterns
    patterns = {
        "SSN": r"\b\d{3}-\d{2}-\d{4}\b",
        "Credit Card": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",
        "Email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    }
    for pii_type, pattern in patterns.items():
        if re.search(pattern, response):
            return f"[REDACTED: Output contained {pii_type}]"
    return response
Rate limiting guardrail
from functools import wraps
import time
def rate_limit(max_calls: int, period: float):
    """Decorator to rate-limit tool calls."""
    calls = []
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [t for t in calls if now - t < period]
            if len(calls) >= max_calls:
                raise RuntimeError(f"Rate limit exceeded: {max_calls} calls per {period}s")
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator
@rate_limit(max_calls=10, period=60.0)
@tool
def limited_api_call(query: str) -> str:
    """An API call with rate limiting."""
    return f"API result for: {query}"