Building AI Agents with LangChain and Claude

I have been building AI applications since the early days of GPT-3. The shift from simple chatbots to autonomous agents is the most exciting evolution I have seen. Agents can reason, use tools, plan multi-step tasks, and even collaborate with other agents.

This post covers how I build production-ready AI agents with LangChain and Claude — from basic tool calling to multi-agent systems with persistent memory.

Why Claude for Agents?

Claude 3.5 Sonnet has become my go-to model for agents because:

• Excellent tool calling — reliable structured outputs with function calling
• Long context window — 200K tokens for complex reasoning chains
• Nuanced instruction following — follows multi-step plans accurately
• Reduced hallucinations — more grounded responses for production use

pip install langchain-anthropic langchain-core langchain-community

Basic Agent with Tool Calling

Start with a simple agent that can call tools:

# agent.py
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# Define tools
@tool
def get_weather(location: str) -> str:
    """Get current weather for a location."""
    # In production, call a real weather API
    weather_data = {
        "jakarta": "Hot and humid, 32°C",
        "singapore": "Partly cloudy, 30°C",
        "tokyo": "Clear skies, 22°C",
    }
    return weather_data.get(location.lower(), f"Weather data not available for {location}")

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # In production, use SerpAPI or Tavily
    return f"Search results for: {query}"

# Initialize LLM
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
)

# Bind tools
tools = [get_weather, search_web]
llm_with_tools = llm.bind_tools(tools)

# Create prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools. Use them when needed."),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# Create agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run
result = agent_executor.invoke({
    "input": "What's the weather in Jakarta and Singapore?"
})

print(result["output"])

Adding Memory

Agents need memory for multi-turn conversations:

# memory.py
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from typing import Dict

# In-memory store (use Redis or database in production)
store: Dict[str, BaseChatMessageHistory] = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Wrap agent with message history
agent_with_chat_history = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# Run with session ID
config = {"configurable": {"session_id": "user-123"}}

response1 = agent_with_chat_history.invoke(
    {"input": "My name is Rahman"},
    config=config,
)

response2 = agent_with_chat_history.invoke(
    {"input": "What's my name?"},
    config=config,
)

print(response2["output"])  # Should remember "Rahman"

Structured Tool Outputs

For production agents, use Pydantic for validated tool outputs:

# structured_tools.py
from pydantic import BaseModel, Field
from typing import Optional
from langchain_core.tools import tool
from datetime import datetime

class WeatherData(BaseModel):
    """Structured weather data."""
    location: str = Field(description="City name")
    temperature: float = Field(description="Temperature in Celsius")
    humidity: int = Field(description="Humidity percentage")
    condition: str = Field(description="Weather condition")
    wind_speed: float = Field(description="Wind speed in km/h")
    timestamp: datetime = Field(default_factory=datetime.now)

class WeatherForecast(BaseModel):
    """Multi-day weather forecast."""
    location: str
    daily_forecasts: list[WeatherData]

@tool(args_schema=WeatherData)
def get_structured_weather(location: str) -> WeatherData:
    """Get structured weather data for a location."""
    # Mock data — replace with real API
    return WeatherData(
        location=location,
        temperature=32.0,
        humidity=75,
        condition="Partly Cloudy",
        wind_speed=12.5,
    )

# Agent can now work with structured data
llm_with_structured = llm.bind_tools([get_structured_weather])

Multi-Agent Systems

For complex tasks, use multiple specialized agents:

# multi_agent.py
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from typing import Literal

# Define specialized tools for each agent
@tool
def analyze_code(code: str) -> str:
    """Analyze code for bugs, security issues, and improvements."""
    return f"Analysis of code (length: {len(code)} chars): No critical issues found."

@tool
def write_tests(code: str) -> str:
    """Generate unit tests for the given code."""
    return f"Generated 5 unit tests for the code."

@tool
def generate_documentation(code: str) -> str:
    """Generate documentation for the given code."""
    return "Generated API documentation with examples."

# Create specialized agents
def create_specialist_agent(
    role: Literal["analyzer", "tester", "documenter"],
    llm: ChatAnthropic,
) -> AgentExecutor:
    tools_by_role = {
        "analyzer": [analyze_code],
        "tester": [write_tests],
        "documenter": [generate_documentation],
    }
    
    prompts_by_role = {
        "analyzer": "You are a code analyst. Analyze code for issues and improvements.",
        "tester": "You are a QA engineer. Write comprehensive tests for code.",
        "documenter": "You are a technical writer. Create clear documentation.",
    }
    
    tools = tools_by_role[role]
    prompt = ChatPromptTemplate.from_messages([
        ("system", prompts_by_role[role]),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    agent = create_tool_calling_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

# Orchestrator agent
@tool
def delegate_to_analyzer(code: str) -> str:
    """Delegate code analysis to the analyzer agent."""
    analyzer = create_specialist_agent("analyzer", llm)
    result = analyzer.invoke({"input": code})
    return result["output"]

@tool
def delegate_to_tester(code: str) -> str:
    """Delegate test generation to the tester agent."""
    tester = create_specialist_agent("tester", llm)
    result = tester.invoke({"input": code})
    return result["output"]

@tool
def delegate_to_documenter(code: str) -> str:
    """Delegate documentation to the documenter agent."""
    documenter = create_specialist_agent("documenter", llm)
    result = documenter.invoke({"input": code})
    return result["output"]

# Main orchestrator
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)

orchestrator_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a code review orchestrator. 
    For any code provided, coordinate analysis, testing, and documentation.
    Use your tools to delegate to specialized agents."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

orchestrator_tools = [delegate_to_analyzer, delegate_to_tester, delegate_to_documenter]
orchestrator_agent = create_tool_calling_agent(llm, orchestrator_tools, orchestrator_prompt)
orchestrator_executor = AgentExecutor(agent=orchestrator_agent, tools=orchestrator_tools, verbose=True)

# Run orchestrated workflow
code_sample = """
def calculate_discount(price: float, discount_percent: float) -> float:
    return price * (1 - discount_percent / 100)
"""

result = orchestrator_executor.invoke({"input": f"Review this code:\n{code_sample}"})
print(result["output"])

Planning and Reasoning

For complex tasks, use chain-of-thought reasoning:

# planning.py
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class PlanStep(BaseModel):
    """A single step in the plan."""
    step_number: int
    action: str
    reasoning: str
    tools_needed: List[str]

class Plan(BaseModel):
    """A multi-step plan for accomplishing a goal."""
    goal: str
    steps: List[PlanStep]
    estimated_complexity: str = Field(description="low, medium, or high")

# Create planning agent
planning_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a planning agent. Given a goal, create a detailed plan.
    For each step, specify:
    1. What action to take
    2. Why this step is needed (reasoning)
    3. What tools or capabilities are required
    
    Think step-by-step and be thorough."""),
    ("human", "Create a plan for: {goal}"),
])

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
structured_llm = llm.with_structured_output(Plan)

planning_chain = planning_prompt | structured_llm

# Generate plan
plan = planning_chain.invoke({
    "goal": "Research and summarize the latest developments in AI agents"
})

print(f"Goal: {plan.goal}")
for step in plan.steps:
    print(f"\nStep {step.step_number}: {step.action}")
    print(f"Reasoning: {step.reasoning}")
    print(f"Tools needed: {', '.join(step.tools_needed)}")

Production Considerations

Error Handling

# error_handling.py
from langchain_core.exceptions import OutputParserException
from langchain_anthropic import ChatAnthropic
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SafeAgentExecutor:
    """Agent executor with comprehensive error handling."""
    
    def __init__(self, agent_executor: AgentExecutor, max_retries: int = 3):
        self.executor = agent_executor
        self.max_retries = max_retries
    
    def safe_invoke(self, input_data: dict, **kwargs):
        for attempt in range(self.max_retries):
            try:
                result = self.executor.invoke(input_data, **kwargs)
                
                # Validate output
                if not result.get("output"):
                    raise ValueError("Empty output from agent")
                
                return {
                    "success": True,
                    "output": result["output"],
                    "attempts": attempt + 1,
                }
                
            except OutputParserException as e:
                logger.warning(f"Parse error on attempt {attempt + 1}: {e}")
                if attempt < self.max_retries - 1:
                    # Retry with error feedback
                    input_data["input"] += f"\n\nPrevious error: {e}\nPlease try again."
                else:
                    return {
                        "success": False,
                        "error": f"Failed to parse output after {self.max_retries} attempts",
                        "details": str(e),
                    }
                    
            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                return {
                    "success": False,
                    "error": "Unexpected error occurred",
                    "details": str(e),
                }
        
        return {
            "success": False,
            "error": "Max retries exceeded",
        }

# Usage
safe_executor = SafeAgentExecutor(agent_executor)
result = safe_executor.safe_invoke({"input": "Analyze the sentiment of this text: 'I love this product!'"})

Token Management

# token_management.py
from langchain_anthropic import ChatAnthropic
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List

class TokenCounter(BaseCallbackHandler):
    """Track token usage across agent runs."""
    
    def __init__(self):
        self.total_tokens = 0
        self.total_cost = 0
        self.calls = []
    
    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        """Track tokens when LLM call ends."""
        if hasattr(response, 'llm_output') and response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            input_tokens = token_usage.get('input_tokens', 0)
            output_tokens = token_usage.get('output_tokens', 0)
            
            # Claude 3.5 Sonnet pricing (as of 2026)
            input_cost = (input_tokens / 1_000_000) * 3.00
            output_cost = (output_tokens / 1_000_000) * 15.00
            
            call_info = {
                'input_tokens': input_tokens,
                'output_tokens': output_tokens,
                'cost': input_cost + output_cost,
            }
            
            self.calls.append(call_info)
            self.total_tokens += input_tokens + output_tokens
            self.total_cost += input_cost + output_cost
    
    def get_summary(self) -> Dict[str, Any]:
        return {
            'total_tokens': self.total_tokens,
            'total_cost': self.total_cost,
            'total_calls': len(self.calls),
            'avg_tokens_per_call': self.total_tokens / len(self.calls) if self.calls else 0,
        }

# Usage
token_counter = TokenCounter()
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
    callbacks=[token_counter],
)

# After agent runs
summary = token_counter.get_summary()
print(f"Total tokens: {summary['total_tokens']}")
print(f"Total cost: ${summary['total_cost']:.4f}")

Streaming Responses

# streaming.py
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Create streaming chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Respond concisely."),
    ("human", "{input}"),
])

llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
    streaming=True,
)

chain = prompt | llm | StrOutputParser()

# Stream output
for chunk in chain.stream({"input": "Explain quantum computing in simple terms"}):
    print(chunk, end="", flush=True)

Real-World Example: Research Agent

Here is a complete research agent I use for my projects:

# research_agent.py
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools.tavily_search import TavilySearchResults
from pydantic import BaseModel
from typing import List
import os

class ResearchReport(BaseModel):
    """Structured research report."""
    topic: str
    key_findings: List[str]
    sources: List[str]
    summary: str
    recommendations: List[str]

@tool
def search_academic_papers(query: str) -> str:
    """Search academic papers on a topic."""
    # Use Semantic Scholar API in production
    return f"Found 10 papers related to: {query}"

@tool
def summarize_content(content: str) -> str:
    """Summarize long content into key points."""
    llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
    response = llm.invoke(f"Summarize the following in bullet points:\n\n{content}")
    return response.content

# Initialize search tool
search_tool = TavilySearchResults(max_results=5)

# Create research agent
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
tools = [search_tool, search_academic_papers, summarize_content]

research_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a research assistant. 
    For any topic:
    1. Search for relevant information
    2. Summarize key findings
    3. Cite sources
    4. Provide actionable recommendations
    
    Be thorough and cite your sources."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

research_agent = create_tool_calling_agent(llm, tools, research_prompt)
research_executor = AgentExecutor(
    agent=research_agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
)

# Run research
report = research_executor.invoke({
    "input": "Research best practices for building AI agents in production"
})

print(report["output"])

Key Takeaways

1. Start with tool calling — Claude’s function calling is reliable for production use
2. Add memory early — Multi-turn context is essential for useful agents
3. Use structured outputs — Pydantic schemas ensure validated, predictable responses
4. Specialize with multi-agent — Complex workflows benefit from specialized agents
5. Plan before acting — Chain-of-thought reasoning improves accuracy
6. Handle errors gracefully — Production agents need robust error handling
7. Track token usage — Monitor costs and optimize prompts
8. Stream for UX — Real-time feedback improves user experience

AI agents are transforming how we build applications. With LangChain and Claude, you can build sophisticated autonomous systems that reason, plan, and execute complex tasks. Start simple, add complexity as needed, and always design for production from day one.

I used similar agent architectures for projects like my OCR Bibliophile app and community platform. The key is matching agent capabilities to the actual problem you are solving.

Building agents is like training a junior developer — give them clear tools, good documentation, and they will surprise you with what they can accomplish.