LangChain Agents & Tools

← Back to LLM Engineering & RAG course

A chain follows a fixed execution path — input goes in, output comes out. An agent is different: it uses the LLM to decide which tools to call, in what order, and whether to call more tools based on intermediate results. This makes agents capable of multi-step reasoning and task execution that no hard-coded chain can match. This lesson covers LangChain's ReAct agent, tool definition, custom tool creation, and the patterns you need to deploy agents reliably in production.

Previous: Lesson 8 — Building a RAG Pipeline from Scratch

The ReAct Reasoning Loop

ReAct (Reasoning + Acting) is the agent pattern used in most LangChain agents. At each step, the LLM produces:

Thought: what it is trying to do and why
Action: which tool to call with what input
Observation: the tool's output

This loop repeats until the LLM decides it has enough information to produce a Final Answer.

text

Question: What is the population of France, and how does that compare to Germany?

Thought: I need to find the population of France.
Action: web_search("current population of France")
Observation: France has a population of approximately 68 million (2025).

Thought: Now I need the population of Germany.
Action: web_search("current population of Germany")
Observation: Germany has a population of approximately 84 million (2025).

Thought: I have both figures. I can now compare them.
Final Answer: France has approximately 68 million people; Germany has approximately 84 million...

Built-In Tools

LangChain provides ready-to-use tools for common integrations:

bash

pip install langchain-community duckduckgo-search wikipedia

python

from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

search = DuckDuckGoSearchRun()
wiki   = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=2))

# Test tools directly
print(search.invoke("LangChain LCEL 2025"))
print(wiki.invoke("transformer architecture neural network"))

Creating a ReAct Agent

python

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain import hub

# Load the standard ReAct prompt from LangChain Hub
prompt = hub.pull("hwchase17/react")

llm   = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [
    DuckDuckGoSearchRun(name="web_search", description="Search the web for current information."),
    WikipediaQueryRun(
        name="wikipedia",
        description="Search Wikipedia for factual background information.",
        api_wrapper=WikipediaAPIWrapper(top_k_results=2)
    )
]

agent = create_react_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,          # prints thought/action/observation loop
    max_iterations=8,      # prevent infinite loops
    handle_parsing_errors=True
)

result = agent_executor.invoke({
    "input": "What Python version was released in October 2024, and what are its headline features?"
})
print(result["output"])

Custom Tool Definition

Any Python function can become an agent tool using the @tool decorator:

python

from langchain_core.tools import tool
from typing import Optional
import requests

@tool
def get_exchange_rate(base: str, target: str) -> str:
    """
    Get the current exchange rate between two currencies.
    Args:
        base:   The base currency code (e.g. USD, GBP, EUR).
        target: The target currency code (e.g. JPY, CAD, CHF).
    Returns the current exchange rate as a string.
    """
    response = requests.get(
        f"https://api.exchangerate-api.com/v4/latest/{base.upper()}"
    )
    data = response.json()
    rate = data["rates"].get(target.upper())
    if rate is None:
        return f"Exchange rate for {base}/{target} not found."
    return f"1 {base.upper()} = {rate} {target.upper()}"

@tool
def calculate(expression: str) -> str:
    """
    Evaluate a mathematical expression and return the result.
    Use standard Python arithmetic syntax: +, -, *, /, **, //, %.
    Example: '(150 * 1.08) / 12'
    """
    try:
        # Restrict to safe arithmetic — no builtins, no imports
        allowed = {k: v for k, v in __builtins__.items()
                   if k in ("abs", "round", "max", "min", "sum")} if isinstance(__builtins__, dict) else {}
        result = eval(expression, {"__builtins__": allowed})
        return str(result)
    except Exception as e:
        return f"Calculation error: {e}"

The docstring is the tool's description — the agent reads it to decide when and how to use the tool. Write clear, specific docstrings.

Structured Tool Input with Pydantic

For tools with multiple parameters, use a BaseModel schema to give the agent a typed, validated input:

python

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
import sqlite3

class DatabaseQueryInput(BaseModel):
    query: str  = Field(description="SQL SELECT query to run against the database")
    limit: int  = Field(default=10, description="Maximum number of rows to return")

def query_database(query: str, limit: int = 10) -> str:
    """Execute a read-only SQL query against the local SQLite database."""
    if not query.strip().upper().startswith("SELECT"):
        return "Error: only SELECT queries are permitted."
    try:
        conn    = sqlite3.connect("app.db")
        cursor  = conn.execute(f"{query} LIMIT {limit}")
        rows    = cursor.fetchall()
        columns = [desc[0] for desc in cursor.description]
        conn.close()
        if not rows:
            return "Query returned no results."
        header = " | ".join(columns)
        lines  = [header, "-" * len(header)]
        lines += [" | ".join(str(v) for v in row) for row in rows]
        return "\n".join(lines)
    except Exception as e:
        return f"Database error: {e}"

db_tool = StructuredTool.from_function(
    func=query_database,
    name="database_query",
    description="Query the application database. Use for questions about users, orders, or products.",
    args_schema=DatabaseQueryInput
)

Agents with Memory

Agents need memory to handle multi-turn conversations, just like chains:

python

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

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

llm   = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_react_agent(llm, tools, conversational_prompt)

executor = AgentExecutor(agent=agent, tools=tools, verbose=False, max_iterations=6)

store: dict = {}

def get_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

agent_with_memory = RunnableWithMessageHistory(
    executor,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

config = {"configurable": {"session_id": "session_1"}}
print(agent_with_memory.invoke({"input": "What is today's GBP/USD rate?"}, config=config))
print(agent_with_memory.invoke({"input": "And what about EUR/USD?"}, config=config))

Combining RAG and Agents

Make the vectorstore a tool so the agent can query private knowledge alongside public web data:

python

from langchain_core.tools import tool
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    collection_name="internal_docs",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db"
)

@tool
def search_internal_docs(query: str) -> str:
    """
    Search the internal company knowledge base for policies, procedures, and documentation.
    Use this before searching the web for any company-specific questions.
    """
    docs = vectorstore.similarity_search(query, k=3)
    if not docs:
        return "No relevant documents found in the internal knowledge base."
    results = []
    for doc in docs:
        source = doc.metadata.get("source", "unknown")
        results.append(f"[{source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(results)

# Add to agent's tool list
tools_with_rag = [search_internal_docs, DuckDuckGoSearchRun(), calculate]

Error Handling and Safety

python

from langchain.agents import AgentExecutor

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,          # hard stop after N iterations
    max_execution_time=30.0,    # hard timeout in seconds
    handle_parsing_errors=True, # recover from malformed tool calls
    early_stopping_method="generate"  # generate answer when limit hit
)

# Wrap in try/except for production use
def safe_agent_invoke(question: str) -> str:
    try:
        result = executor.invoke({"input": question})
        return result.get("output", "No answer produced.")
    except Exception as e:
        return f"Agent error: {type(e).__name__}: {e}"