Beginner Project: Q&A Agent with Tool Use and Routing

Beginner⏱45 minutes📖3 min readLesson 10 of 10

Beginner Project: Q&A Agent with Tool Use and Routing

Build a complete Q&A agent using LangGraph with tool use, conditional routing, and state management.

In this capstone project, you'll build a complete Q&A agent that can answer questions using tools and intelligent routing. This brings together everything you've learned: state, nodes, edges, LLMs, tools, and conditional branching.

Project Overview

The agent will:

Classify the user's question into a category
Route to the appropriate handler based on the category
Execute tools or generate responses as needed
Loop if tools were called, to process results
Return a final answer

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Step 1: Setup and Imports

python

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolExecutor
from typing_extensions import TypedDict, List, Annotated
from typing import Any
from operator import add
import json

Step 2: Define Tools

python

@tool
def web_search(query: str) -> str:
    """Search the web for current information. Use for news, facts, and general knowledge."""
    # In production, call a real search API
    return f"## Web Results for '{query}'\n"
           f"1. LangGraph is a framework for building stateful AI agents.\n"
           f"2. It was created by LangChain Inc.\n"
           f"3. Version 0.2 was released in 2024."

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression. Use Python syntax: +, -, *, /, **, //, %."""
    try:
        safe_dict = {"__builtins__": {}}
        result = eval(expression, safe_dict)
        return str(result)
    except Exception as e:
        return f"Calculation error: {e}"

@tool
def get_timezone(city: str) -> str:
    """Get the timezone for a given city."""
    timezones = {
        "new york": "America/New_York (UTC-5)",
        "london": "Europe/London (UTC+0)",
        "tokyo": "Asia/Tokyo (UTC+9)",
        "paris": "Europe/Paris (UTC+1)",
        "sydney": "Australia/Sydney (UTC+11)"
    }
    return timezones.get(city.lower(), f"Timezone not found for {city}")

ℹ️Note

These are simulated tools for the project. In production, replace them with real API calls to search engines, calculators, and timezone databases.

Step 3: Initialize LLM and Tool Bindings

python

llm = ChatOpenAI(model="gpt-4o-mini")

all_tools = [web_search, calculator, get_timezone]
llm_with_tools = llm.bind_tools(all_tools)
tool_executor = ToolExecutor(all_tools)

Step 4: Define State

python

class AgentState(TypedDict):
    messages: Annotated[List[Any], add]
    question: str
    category: str
    tool_results: List[str]
    final_answer: str
    error: str

Field	Type	Purpose
`messages`	`Annotated[List, add]`	Full conversation history with appending
`question`	`str`	The user's original question
`category`	`str`	Classification result for routing
`tool_results`	`List[str]`	Accumulated tool outputs
`final_answer`	`str`	The final response to the user
`error`	`str`	Error tracking

Step 5: Define Nodes

Classify Node

python

def classify_question(state: AgentState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Classify the question into exactly one category:\n"
                   "- 'web_search': questions about news, facts, people, concepts\n"
                   "- 'calculator': math problems, calculations, equations\n"
                   "- 'chat': general conversation, opinions, explanations\n\n"
                   "Respond with ONLY the category name."),
        ("human", "{question}")
    ])
    chain = prompt | llm | StrOutputParser()
    category = chain.invoke({"question": state["question"]}).strip().lower()

    if category not in ["web_search", "calculator", "chat"]:
        category = "chat"  # Default fallback

    return {"category": category, "messages": [HumanMessage(state["question"])]}

Web Search Node

python

def web_search_node(state: AgentState) -> dict:
    try:
        result = tool_executor.invoke({
            "name": "web_search",
            "args": {"query": state["question"]},
            "id": "search_1",
            "type": "tool_call"
        })
        return {
            "tool_results": state["tool_results"] + [str(result)],
            "messages": [ToolMessage(content=str(result), tool_call_id="search_1")]
        }
    except Exception as e:
        return {"error": f"Web search failed: {e}"}

Calculator Node

python

def calculator_node(state: AgentState) -> dict:
    try:
        result = tool_executor.invoke({
            "name": "calculator",
            "args": {"expression": state["question"]},
            "id": "calc_1",
            "type": "tool_call"
        })
        return {
            "tool_results": state["tool_results"] + [str(result)],
            "messages": [ToolMessage(content=str(result), tool_call_id="calc_1")]
        }
    except Exception as e:
        return {"error": f"Calculator failed: {e}"}

Agent Node (Handles Final Response)

python

def agent_node(state: AgentState) -> dict:
    context = "\n".join(state["tool_results"]) if state["tool_results"] else "No tools were needed."

    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful Q&A assistant. Answer the user's question "
                   "based on the available context. Be concise and accurate.\n\n"
                   "Context:\n{context}"),
        ("human", "{question}")
    ])

    chain = prompt | llm | StrOutputParser()
    answer = chain.invoke({
        "context": context,
        "question": state["question"]
    })

    return {
        "final_answer": answer,
        "messages": [AIMessage(content=answer)]
    }

💡Tip

The agent node reads tool_results from state. If tools were used, it synthesizes a response from their output. If not, it answers directly.

Step 6: Define Router Functions

python

def route_by_category(state: AgentState) -> str:
    return state["category"]

def should_continue(state: AgentState) -> str:
    if state.get("error"):
        return "error"
    if state["category"] in ("web_search", "calculator"):
        return "use_agent"  # Tool was called, now synthesize
    return "done"  # Chat doesn't need tools

Step 7: Build the Graph

python

builder = StateGraph(AgentState)

# Add nodes
builder.add_node("classify", classify_question)
builder.add_node("web_search", web_search_node)
builder.add_node("calculator", calculator_node)
builder.add_node("agent", agent_node)

# Add edges
builder.add_edge(START, "classify")

# Conditional routing from classifier
builder.add_conditional_edges(
    "classify",
    route_by_category,
    {
        "web_search": "web_search",
        "calculator": "calculator",
        "chat": "agent"  # Chat goes directly to agent
    }
)

# After tools, go to agent for final answer
builder.add_edge("web_search", "agent")
builder.add_edge("calculator", "agent")
builder.add_edge("agent", END)

# Compile
app = builder.compile()

Step 8: Run the Agent

python

def ask_question(question: str) -> str:
    result = app.invoke({
        "messages": [],
        "question": question,
        "category": "",
        "tool_results": [],
        "final_answer": "",
        "error": ""
    })
    return result["final_answer"]

# Test the agent
print(ask_question("What is LangGraph?"))
# → Uses web_search and returns answer from context

print(ask_question("What is 245 * 37?"))
# → Routes to calculator, returns result via agent

print(ask_question("What timezone is Tokyo in?"))
# → Uses web_search (keyword: timezone), returns answer

print(ask_question("What is the meaning of life?"))
# → Routes to chat, LLM answers directly

print(ask_question("Solve: (15 + 7) * 2"))
# → Routes to calculator, evaluates expression

✅Success

Your agent now intelligently routes questions to the right tool and synthesizes a final answer. This is the core pattern behind all LangGraph agents.

Step 9: Add Streaming

python

def ask_with_streaming(question: str) -> None:
    print(f"\nQ: {question}")
    print("-" * 40)

    for event in app.stream({
        "messages": [],
        "question": question,
        "category": "",
        "tool_results": [],
        "final_answer": "",
        "error": ""
    }):
        for node, update in event.items():
            if node == "__end__":
                continue
            if "category" in update:
                print(f"[Classified as: {update['category']}]")
            if "tool_results" in update:
                for r in update["tool_results"]:
                    print(f"[Tool result]: {r[:100]}...")
            if "final_answer" in update:
                print(f"[Answer]: {update['final_answer']}")

Step 10: Error Handling and Edge Cases

python

def safe_ask(question: str) -> str:
    try:
        result = app.invoke(
            {
                "messages": [],
                "question": question,
                "category": "",
                "tool_results": [],
                "final_answer": "",
                "error": ""
            },
            {"recursion_limit": 10}
        )

        if result.get("error"):
            return f"An error occurred: {result['error']}"

        return result.get("final_answer", "No answer generated.")

    except Exception as e:
        return f"Sorry, I encountered an error: {str(e)}"

# Test edge cases
print(safe_ask(""))                         # Empty question
print(safe_ask("123456" * 1000))            # Very long question
print(safe_ask("∫ x² dx"))                  # Unsupported input

ℹ️Note

Always wrap app.invoke() in try/except and set recursion_limit. This prevents crashes from unexpected inputs and infinite loops.

Project Extension Ideas

Add more tools: Integrate a real search API (Tavily, SerpAPI), database query tool, or file reader
Add memory: Use the add_messages reducer on messages to track conversation history across turns
Add confidence scoring: Make the classifier return a confidence score, route to fallback if low
Parallel tool calls: For questions that need multiple tools, call them in parallel
Human handoff: Add an interrupt for questions the agent cannot answer

Complete Project Code

python

# Full project — combine all steps above
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolExecutor
from typing_extensions import TypedDict, List, Annotated
from typing import Any
from operator import add

# Tools
@tool
def web_search(query: str) -> str:
    """Search the web for current information."""
    return f"Web results for '{query}' — LangGraph is a stateful agent framework."

@tool
def calculator(expression: str) -> str:
    """Evaluate mathematical expressions."""
    return str(eval(expression, {"__builtins__": {}}, {}))

# Setup
llm = ChatOpenAI(model="gpt-4o-mini")
all_tools = [web_search, calculator]
tool_executor = ToolExecutor(all_tools)

# State
class AgentState(TypedDict):
    messages: Annotated[List[Any], add]
    question: str
    category: str
    tool_results: List[str]
    final_answer: str
    error: str

# Nodes
def classify_question(state: AgentState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Classify as 'web_search', 'calculator', or 'chat'. Respond with only the category."),
        ("human", "{question}")
    ])
    chain = prompt | llm | StrOutputParser()
    cat = chain.invoke({"question": state["question"]}).strip().lower()
    return {"category": cat if cat in ("web_search", "calculator", "chat") else "chat",
            "messages": [HumanMessage(state["question"])]}

def web_search_node(state: AgentState) -> dict:
    result = tool_executor.invoke({"name": "web_search", "args": {"query": state["question"]}, "id": "s1", "type": "tool_call"})
    return {"tool_results": state["tool_results"] + [str(result)],
            "messages": [ToolMessage(content=str(result), tool_call_id="s1")]}

def calculator_node(state: AgentState) -> dict:
    result = tool_executor.invoke({"name": "calculator", "args": {"expression": state["question"]}, "id": "c1", "type": "tool_call"})
    return {"tool_results": state["tool_results"] + [str(result)],
            "messages": [ToolMessage(content=str(result), tool_call_id="c1")]}

def agent_node(state: AgentState) -> dict:
    context = "\n".join(state["tool_results"]) if state["tool_results"] else "No tools needed."
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer concisely using context:\n{context}"),
        ("human", "{question}")
    ])
    answer = (prompt | llm | StrOutputParser()).invoke({"context": context, "question": state["question"]})
    return {"final_answer": answer, "messages": [AIMessage(content=answer)]}

# Router
def route_by_category(state: AgentState) -> str:
    return state["category"]

# Graph
builder = StateGraph(AgentState)
builder.add_node("classify", classify_question)
builder.add_node("web_search", web_search_node)
builder.add_node("calculator", calculator_node)
builder.add_node("agent", agent_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_by_category, {
    "web_search": "web_search",
    "calculator": "calculator",
    "chat": "agent"
})
builder.add_edge("web_search", "agent")
builder.add_edge("calculator", "agent")
builder.add_edge("agent", END)

app = builder.compile()

# Run
if __name__ == "__main__":
    questions = [
        "What is LangGraph?",
        "What is 15 * 37?",
        "What timezone is Tokyo in?",
        "Tell me a fun fact"
    ]
    for q in questions:
        result = app.invoke({"messages": [], "question": q, "category": "", "tool_results": [], "final_answer": "", "error": ""})
        print(f"Q: {q}\nA: {result['final_answer']}\n")