Build a LangChain Agent with Memory and Tools (Full Example)

pip install langchain langchain-openai langchain-community \
    duckduckgo-search fastapi uvicorn python-dotenv

# .env
OPENAI_API_KEY=your_key_here
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your_langsmith_key   # optional but recommended
LANGCHAIN_PROJECT=aria-agent

import os
import math
from datetime import datetime
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun

load_dotenv()

# Tool 1: Web search
search = DuckDuckGoSearchRun()

@tool
def search_web(query: str) -> str:
    """Search the internet for current information about a topic, person, or event.
    Use this when you need facts that might have changed recently or are not in your training data.
    Input: a search query string.
    """
    try:
        result = search.run(query)
        return result if result else "No results found for this query."
    except Exception as e:
        return f"Search failed: {str(e)}. Try a different query."

# Tool 2: Calculator
@tool
def calculate(expression: str) -> str:
    """Perform mathematical calculations.
    Input: a valid Python math expression as a string (e.g., '15 * 1.08', 'math.sqrt(144)', '(100 - 20) / 4').
    Do NOT include variable names — only numbers and operators.
    """
    try:
        # Restrict to safe math operations
        safe_globals = {"math": math, "__builtins__": {}}
        result = eval(expression, safe_globals)
        return f"Result: {result}"
    except ZeroDivisionError:
        return "Error: Division by zero"
    except Exception as e:
        return f"Calculation error: {str(e)}. Check the expression format."

# Tool 3: Product catalog lookup
PRODUCT_CATALOG = {
    "python course": {
        "name": "Python Complete Course",
        "price": 149.99,
        "description": "40-hour comprehensive Python course from basics to advanced topics",
        "rating": 4.8,
        "in_stock": True,
    },
    "ai toolkit": {
        "name": "AI Toolkit Pro",
        "price": 299.99,
        "description": "50 pre-built AI agent templates with customization guides",
        "rating": 4.6,
        "in_stock": True,
    },
    "data science bundle": {
        "name": "Data Science Bundle",
        "price": 199.99,
        "description": "Complete data science curriculum: NumPy, Pandas, Scikit-learn, visualization",
        "rating": 4.7,
        "in_stock": False,
    },
    "ml fundamentals": {
        "name": "Machine Learning Fundamentals",
        "price": 129.99,
        "description": "Math and implementation foundations of machine learning algorithms",
        "rating": 4.5,
        "in_stock": True,
    },
}

@tool
def lookup_product(product_name: str) -> str:
    """Look up information about a product in our catalog.
    Use this when a user asks about specific products, their prices, availability, or descriptions.
    Input: the product name or a close approximation (e.g., 'python course', 'AI toolkit').
    """
    query = product_name.lower().strip()

    # Direct match
    if query in PRODUCT_CATALOG:
        p = PRODUCT_CATALOG[query]
        stock_status = "In Stock" if p["in_stock"] else "Currently Out of Stock"
        return (f"Product: {p['name']}\n"
                f"Price: ${p['price']}\n"
                f"Rating: {p['rating']}/5.0\n"
                f"Status: {stock_status}\n"
                f"Description: {p['description']}")

    # Fuzzy match
    for key, product in PRODUCT_CATALOG.items():
        if query in key or key in query:
            p = product
            stock_status = "In Stock" if p["in_stock"] else "Currently Out of Stock"
            return (f"Product: {p['name']}\n"
                    f"Price: ${p['price']}\n"
                    f"Rating: {p['rating']}/5.0\n"
                    f"Status: {stock_status}\n"
                    f"Description: {p['description']}")

    available = ", ".join(PRODUCT_CATALOG.keys())
    return f"Product '{product_name}' not found. Available products: {available}"

# Tool 4: Date and time
@tool
def get_current_datetime(timezone: str = "UTC") -> str:
    """Get the current date and time.
    Use this when the user asks about the current time, date, or wants to know what day it is.
    Input: timezone name (default: 'UTC'). Examples: 'UTC', 'US/Eastern', 'Europe/London'.
    """
    try:
        from zoneinfo import ZoneInfo
        tz = ZoneInfo(timezone)
        now = datetime.now(tz)
        return f"Current datetime ({timezone}): {now.strftime('%A, %B %d, %Y at %I:%M %p %Z')}"
    except Exception:
        now = datetime.utcnow()
        return f"Current datetime (UTC): {now.strftime('%A, %B %d, %Y at %I:%M %p UTC')}"

# Tool 5: Unit converter
@tool
def convert_units(value: float, from_unit: str, to_unit: str) -> str:
    """Convert between common units of measurement.
    Supports: temperature (celsius/fahrenheit/kelvin), length (meters/feet/inches/km/miles),
    weight (kg/lbs/grams/ounces).
    Input: numeric value, source unit, target unit.
    """
    from_unit = from_unit.lower().strip()
    to_unit = to_unit.lower().strip()

    conversions = {
        ("celsius", "fahrenheit"): lambda v: v * 9/5 + 32,
        ("fahrenheit", "celsius"): lambda v: (v - 32) * 5/9,
        ("celsius", "kelvin"): lambda v: v + 273.15,
        ("kelvin", "celsius"): lambda v: v - 273.15,
        ("meters", "feet"): lambda v: v * 3.28084,
        ("feet", "meters"): lambda v: v / 3.28084,
        ("km", "miles"): lambda v: v * 0.621371,
        ("miles", "km"): lambda v: v / 0.621371,
        ("kg", "lbs"): lambda v: v * 2.20462,
        ("lbs", "kg"): lambda v: v / 2.20462,
    }

    key = (from_unit, to_unit)
    if key in conversions:
        result = conversions[key](value)
        return f"{value} {from_unit} = {result:.4f} {to_unit}"

    return f"Conversion from {from_unit} to {to_unit} is not supported."

# Collect all tools
tools = [search_web, calculate, lookup_product, get_current_datetime, convert_units]

print(f"Tools loaded: {[t.name for t in tools]}")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import create_openai_tools_agent, AgentExecutor

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# System prompt that defines Aria's personality and behavior
system_prompt = """You are Aria, a helpful AI assistant for AiTechWorlds.com.

Your capabilities:
- Search the web for current information
- Perform calculations
- Look up product catalog information
- Get the current date and time
- Convert between units

Guidelines:
- Be conversational and remember context from earlier in our conversation
- Use tools when you need current or specific information
- For calculations, always use the calculator tool rather than computing in your head
- When looking up products, always check the catalog first before making up information
- If a tool fails, explain what happened and suggest alternatives
- Keep responses focused and concise — don't pad with unnecessary text

Current date context: You were last updated in early 2026. Use search_web for anything after that."""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_tools_agent(
    llm=llm,
    tools=tools,
    prompt=prompt,
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # shows the full reasoning trace
    max_iterations=10,      # prevent infinite loops
    handle_parsing_errors=True,
    return_intermediate_steps=True,  # capture the tool call trace
)

print("Agent created successfully")

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory

# In-memory store — swap for Redis in production
session_store: dict[str, BaseChatMessageHistory] = {}

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

# Wrap agent executor with memory management
agent_with_memory = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

print("Memory wrapper added")

def chat(message: str, session_id: str = "demo_session", show_trace: bool = True):
    """Send a message to the agent and show the response with optional trace."""
    config = {"configurable": {"session_id": session_id}}

    result = agent_with_memory.invoke(
        {"input": message},
        config=config
    )

    print(f"\n{'='*65}")
    print(f"USER: {message}")

    if show_trace and "intermediate_steps" in result:
        steps = result["intermediate_steps"]
        if steps:
            print(f"\n[TRACE — {len(steps)} step(s)]")
            for i, (action, observation) in enumerate(steps):
                print(f"  Step {i+1}: Tool '{action.tool}'")
                print(f"    Input: {str(action.tool_input)[:100]}")
                print(f"    Output: {str(observation)[:120]}")

    print(f"\nARIA: {result['output']}")
    return result["output"]

# Run the full conversation
print("Starting multi-turn conversation with Aria...\n")

# Turn 1: Simple greeting with personal info
chat("Hi Aria! I'm Marcus, and I'm interested in learning Python. What do you recommend?",
     session_id="marcus_001")

# Turn 2: Memory test — agent should remember the user's name and interest
chat("What's my name and what was I asking about?",
     session_id="marcus_001")

# Turn 3: Tool use — product lookup
chat("How much does the Python course cost, and is it in stock?",
     session_id="marcus_001")

# Turn 4: Tool use — calculation
chat("If I buy the Python course and the AI Toolkit together, and I have a 15% discount, what's the total?",
     session_id="marcus_001")

# Turn 5: Tool use — web search for current info
chat("What are the latest Python 3.14 features? I want to know what's new.",
     session_id="marcus_001")

# Turn 6: Tool use — date/time
chat("What day is it today?",
     session_id="marcus_001")

# Turn 7: Context recall combining memory + tool use
chat("Based on everything we've discussed, summarize what I should do this week to start my Python journey.",
     session_id="marcus_001")

> Entering new AgentExecutor chain...
Invoking: `lookup_product` with `{'product_name': 'python course'}`
Product: Python Complete Course
Price: $149.99
...

Invoking: `lookup_product` with `{'product_name': 'ai toolkit'}`
Product: AI Toolkit Pro
Price: $299.99
...

Invoking: `calculate` with `{'expression': '(149.99 + 299.99) * (1 - 0.15)'}`
Result: 382.482

> Finished chain.
ARIA: With a 15% discount on both the Python Complete Course ($149.99) 
and the AI Toolkit Pro ($299.99), your total comes to $382.48.
Here is the breakdown:
- Combined original price: $449.98
- Discount (15%): $67.50
- Final total: $382.48

# aria_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="Aria Agent API", version="1.0")

class ChatRequest(BaseModel):
    session_id: str
    message: str
    show_trace: bool = False

class TraceStep(BaseModel):
    tool: str
    input: str
    output: str

class ChatResponse(BaseModel):
    session_id: str
    response: str
    trace: list[TraceStep] = []
    turn_count: int

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    if not request.message.strip():
        raise HTTPException(status_code=400, detail="Message cannot be empty")

    config = {"configurable": {"session_id": request.session_id}}

    try:
        result = await agent_with_memory.ainvoke(
            {"input": request.message},
            config=config
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent error: {str(e)}")

    trace_steps = []
    if request.show_trace and "intermediate_steps" in result:
        for action, observation in result["intermediate_steps"]:
            trace_steps.append(TraceStep(
                tool=action.tool,
                input=str(action.tool_input)[:200],
                output=str(observation)[:300],
            ))

    history = get_session_history(request.session_id)
    turn_count = len(history.messages) // 2

    return ChatResponse(
        session_id=request.session_id,
        response=result["output"],
        trace=trace_steps,
        turn_count=turn_count,
    )

@app.delete("/sessions/{session_id}")
async def clear_session(session_id: str):
    if session_id in session_store:
        session_store[session_id].clear()
        return {"message": f"Session {session_id} cleared"}
    raise HTTPException(status_code=404, detail="Session not found")

@app.get("/sessions/{session_id}/history")
async def get_history(session_id: str):
    if session_id not in session_store:
        raise HTTPException(status_code=404, detail="Session not found")

    messages = session_store[session_id].messages
    return {
        "session_id": session_id,
        "turn_count": len(messages) // 2,
        "messages": [
            {"role": msg.type, "content": msg.content[:200]}
            for msg in messages
        ]
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8002)

# Start the server
uvicorn aria_api:app --reload --port 8002

# First turn
curl -X POST http://localhost:8002/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id": "marcus_001", "message": "Hi, I am Marcus. What products do you have?", "show_trace": true}'

# Second turn — tests memory
curl -X POST http://localhost:8002/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id": "marcus_001", "message": "What was I asking about earlier?", "show_trace": false}'

def add_tool_to_agent(new_tool, session_id: str = None):
    """Add a new tool to the agent dynamically."""
    global tools, agent, agent_executor, agent_with_memory

    tools.append(new_tool)

    # Rebuild the agent with the updated tools list
    agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt)
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=10,
        handle_parsing_errors=True,
        return_intermediate_steps=True,
    )
    agent_with_memory = RunnableWithMessageHistory(
        agent_executor,
        get_session_history,
        input_messages_key="input",
        history_messages_key="chat_history",
    )
    print(f"Tool '{new_tool.name}' added. Current tools: {[t.name for t in tools]}")

# Example: add a new tool dynamically
@tool
def get_joke(topic: str = "general") -> str:
    """Tell a short, clean joke on a given topic. Use when user wants a joke or some humor."""
    jokes = {
        "programming": "Why do programmers prefer dark mode? Because light attracts bugs.",
        "python": "Why is Python so popular? Because the syntax is clear — no curly brace drama.",
        "general": "Why don't scientists trust atoms? Because they make up everything.",
    }
    return jokes.get(topic.lower(), jokes["general"])

add_tool_to_agent(get_joke)

from langchain_core.callbacks import BaseCallbackHandler

class AgentErrorHandler(BaseCallbackHandler):
    """Catches and logs agent errors gracefully."""

    def on_tool_error(self, error, **kwargs):
        print(f"[WARNING] Tool error: {error}")
        # In production, send this to your error tracking (Sentry, etc.)

    def on_agent_action(self, action, **kwargs):
        # Log each tool call for monitoring
        print(f"[AGENT] Calling tool: {action.tool}")

    def on_agent_finish(self, finish, **kwargs):
        # Log successful completions
        print(f"[AGENT] Finished: {finish.return_values.get('output', '')[:100]}")

error_handler = AgentErrorHandler()

# Rebuild with error handler
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=False,   # use the callback handler instead
    max_iterations=10,
    handle_parsing_errors=True,
    return_intermediate_steps=True,
    callbacks=[error_handler],
)

User Message
    ↓
FastAPI Endpoint
    ↓
RunnableWithMessageHistory
    ├── Gets session history from store
    └── Injects history into agent input
         ↓
    AgentExecutor
         ├── ChatPromptTemplate (system + history + input + scratchpad)
         ├── GPT-4o (decides: respond directly OR call tool)
         └── Tool Loop:
              ├── search_web → DuckDuckGo results
              ├── calculate → Python eval result
              ├── lookup_product → catalog response
              ├── get_current_datetime → formatted time
              └── convert_units → conversion result
         ↓
    Final Response
    ↓
Session history updated
    ↓
Response returned to user