Langchain Agents

langchain-agents (Python)

Overview

Agents combine language models with tools to create systems that can reason about tasks, decide which tools to use, and iteratively work towards solutions. The create_agent() function provides a production-ready agent implementation built on LangGraph.

Key Concepts:

• Agent Loop: The model decides → calls tools → observes results → repeats until done
• ReAct Pattern: Reasoning and Acting - the agent reasons about what to do, then acts by calling tools
• Graph-based Runtime: Agents run on a LangGraph graph with nodes (model, tools, middleware) and edges

When to Use Agents

Decision Tables

Choosing Agent Configuration

Tool Strategy

Code Examples

Basic Agent with Tools

from langchain.agents import create_agent
from langchain.tools import tool

# Define tools using @tool decorator
@tool
def search(query: str) -> str:
    """Search for information on the web.
    
    Args:
        query: The search query
    """
    # Your search implementation
    return f"Results for: {query}"

@tool
def get_weather(location: str) -> str:
    """Get current weather for a location.
    
    Args:
        location: City name
    """
    return f"Weather in {location}: Sunny, 72°F"

# Create agent
agent = create_agent(
    model="gpt-4.1",
    tools=[search, get_weather],
)

# Invoke agent
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "What's the weather in San Francisco?"}
    ]
})

print(result["messages"][-1].content)

Agent with System Prompt

from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4.1",
    tools=[search, calculator],
    system_prompt="""You are a helpful research assistant.
Always cite your sources when using the search tool.
Show your work when performing calculations.""",
)

Agent Loop Execution Flow

# The agent runs in a loop:
# 1. Model receives user message
# 2. Model decides to call a tool (or finish)
# 3. Tool executes and returns result
# 4. Result goes back to model
# 5. Repeat until model decides to finish

agent = create_agent(
    model="gpt-4.1",
    tools=[search, get_weather],
)

# This single invoke() call handles the entire loop
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Search for the capital of France, then get its weather"
    }]
})

# Agent automatically:
# - Calls search tool for capital
# - Receives "Paris"
# - Calls weather tool for Paris
# - Receives weather data
# - Responds with final answer

Streaming Agent Progress

from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4.1",
    tools=[search],
)

# Stream with updates mode to see each step
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Search for LangChain"}]},
    stream_mode=["updates"],
):
    print(f"Step: {chunk}")

# Stream with messages mode for LLM tokens
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Search for LangChain"}]},
    stream_mode=["messages"],
):
    token, metadata = chunk
    if token.content:
        print(token.content, end="", flush=True)

Agent with Persistence

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

agent = create_agent(
    model="gpt-4.1",
    tools=[search],
    checkpointer=checkpointer,
)

# First conversation
config = {"configurable": {"thread_id": "user-123"}}
agent.invoke({
    "messages": [{"role": "user", "content": "My name is Alice"}]
}, config=config)

# Later conversation - agent remembers
result = agent.invoke({
    "messages": [{"role": "user", "content": "What's my name?"}]
}, config=config)
# Response: "Your name is Alice"

Multiple Tool Calls in Parallel

# Models can call multiple tools simultaneously
agent = create_agent(
    model="gpt-4.1",
    tools=[get_weather, get_news],
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Get weather for NYC and latest news for SF"
    }]
})

# Agent may call both tools in parallel in a single step

Dynamic Tools (Runtime-Dependent)

from langchain.agents import create_agent

def get_tools(state):
    """Tools can depend on current state."""
    user_id = state.get("config", {}).get("configurable", {}).get("user_id")
    return [
        get_user_specific_tool(user_id),
        common_tool,
    ]

agent = create_agent(
    model="gpt-4.1",
    tools=get_tools,  # Pass function instead of list
)

Error Handling in Agents

from langchain.agents import create_agent, wrap_tool_call

# Custom error handling middleware
@wrap_tool_call
async def error_handler(tool_call, handler):
    try:
        return await handler(tool_call)
    except Exception as error:
        return {
            **tool_call,
            "content": f"Tool error: {str(error)}",
        }

agent = create_agent(
    model="gpt-4.1",
    tools=[risky_tool],
    middleware=[error_handler],
)

Tool with Type Hints

from langchain.tools import tool
from typing import Literal

@tool
def calculate(
    operation: Literal["add", "subtract", "multiply", "divide"],
    a: float,
    b: float,
) -> float:
    """Perform a mathematical calculation.
    
    Args:
        operation: The operation to perform
        a: First number
        b: Second number
    """
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        return a / b

Boundaries

What Agents CAN Configure

✅ Model: Any chat model (OpenAI, Anthropic, Google, etc.) ✅ Tools: Custom tools, built-in tools, dynamic tools ✅ System Prompt: Instructions for agent behavior ✅ Middleware: Human-in-the-loop, error handling, logging ✅ Checkpointer: Memory/persistence across conversations ✅ Response Format: Structured output schemas (Pydantic, TypedDict, JSON Schema) ✅ Max Iterations: Prevent infinite loops

What Agents CANNOT Configure

❌ Direct Graph Structure: Use LangGraph directly for custom flows ❌ Tool Execution Order: Model decides which tools to call ❌ Interrupt Model Decision: Can only interrupt before tool execution ❌ Multiple Models: One agent = one model (use subagents for multiple)

Gotchas

1. Agent Doesn't Stop (Infinite Loop)

# ❌ Problem: No clear stopping condition
agent = create_agent(
    model="gpt-4.1",
    tools=[search],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Keep searching until perfect"}]
})

# ✅ Solution: Set max iterations
agent = create_agent(
    model="gpt-4.1",
    tools=[search],
    max_iterations=10,  # Stop after 10 tool calls
)

2. Tool Not Being Called

# ❌ Problem: Vague tool description
@tool
def bad_tool(input: str) -> str:
    """Does stuff."""  # Too vague!
    return "result"

# ✅ Solution: Clear, specific descriptions
@tool
def web_search(query: str) -> str:
    """Search the web for current information about a topic.
    
    Use this when you need recent data that wasn't in your training.
    
    Args:
        query: The search query (2-10 words)
    """
    return "result"

3. State Not Persisting

# ❌ Problem: No checkpointer
agent = create_agent(
    model="gpt-4.1",
    tools=[search],
)

# Each invoke is isolated - no memory
agent.invoke({"messages": [{"role": "user", "content": "Hi, I'm Bob"}]})
agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]})
# Agent doesn't remember "Bob"

# ✅ Solution: Add checkpointer and thread_id
from langgraph.checkpoint.memory import MemorySaver

agent = create_agent(
    model="gpt-4.1",
    tools=[search],
    checkpointer=MemorySaver(),
)

config = {"configurable": {"thread_id": "session-1"}}
agent.invoke({"messages": [{"role": "user", "content": "Hi, I'm Bob"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]}, config=config)
# Agent remembers: "Your name is Bob"

4. Messages vs State Confusion

# Agent state includes more than just messages
result = agent.invoke({
    "messages": [{"role": "user", "content": "Hello"}]
})

# ✅ Access full conversation history
print(result["messages"])  # List of all messages

# ✅ Access structured output (if configured)
print(result.get("structured_response"))

# ❌ Don't try to access result.content directly
# print(result.content)  # KeyError!

5. Tool Results Must Be Serializable

from datetime import datetime

# ❌ Problem: Returning non-serializable objects
@tool
def bad_get_time() -> datetime:
    """Get current time."""
    return datetime.now()  # datetime objects need special handling

# ✅ Solution: Return serializable data
@tool
def good_get_time() -> str:
    """Get current time."""
    return datetime.now().isoformat()  # String is serializable

6. Streaming Modes Matter

# Different stream modes show different information

# "values" - Full state after each step
for mode, chunk in agent.stream(input, stream_mode=["values"]):
    print(chunk["messages"])  # All messages so far

# "updates" - Only what changed in each step
for mode, chunk in agent.stream(input, stream_mode=["updates"]):
    print(chunk)  # Just the delta

# "messages" - LLM token stream
for mode, chunk in agent.stream(input, stream_mode=["messages"]):
    token, metadata = chunk
    print(token.content, end="", flush=True)

7. Async Tools Must Be Awaited Properly

from langchain.tools import tool

# ✅ Async tool properly defined
@tool
async def async_search(query: str) -> str:
    """Async search tool."""
    # Use await for async operations
    result = await some_async_api_call(query)
    return result

# Agent handles both sync and async tools
agent = create_agent(
    model="gpt-4.1",
    tools=[sync_tool, async_search],
)

# Use ainvoke for async execution
result = await agent.ainvoke({
    "messages": [{"role": "user", "content": "Search something"}]
})

Links to Documentation

• Agents Overview
• create_agent API Reference
• LangGraph Concepts
• Tool Calling Guide
• Streaming Guide
• Human-in-the-Loop