Building AI Agents with LangChain

# Install core LangChain packages
pip install langchain langchain-core langchain-community
# Install OpenAI integration (or use any other LLM provider)
pip install langchain-openai
# Install additional tools used in this guide
pip install duckduckgo-search wikipedia numexpr
# Optional: LangSmith for tracing and debugging
pip install langsmith

import os
# Set your OpenAI API key
os.environ['OPENAI_API_KEY'] = 'your-openai-api-key-here'
# Optional: LangSmith tracing (highly recommended for debugging)
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY']    = 'your-langsmith-api-key'
os.environ['LANGCHAIN_PROJECT']    = 'langchain-agent-demo'

from langchain_openai import ChatOpenAI
# Initialise the LLM
llm = ChatOpenAI(
    model       = 'gpt-4o',   # or 'gpt-3.5-turbo', 'gpt-4-turbo'
    temperature = 0,           # 0 = deterministic; higher = more creative
    max_tokens  = 1000
)
# Simple invocation
response = llm.invoke('What is LangChain?')
print(response.content)

from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
    ('system', 'You are a helpful assistant specialising in {domain}.'),
    ('human',  'Answer this question: {question}'),
])
# Format the prompt with values
formatted = prompt.format_messages(
    domain   = 'data engineering',
    question = 'What is Apache Spark?'
)
print(formatted)

from langchain_core.output_parsers import StrOutputParser
# Build a chain: prompt | llm | output parser
chain = prompt | llm | StrOutputParser()
# Invoke the chain
result = chain.invoke({
    'domain':   'data engineering',
    'question': 'What is Apache Spark?'
})
print(result)  # Returns plain string, not a Message object

from langchain_core.tools import tool
# The docstring is what the LLM reads to decide when to use this tool
@tool
def calculate_compound_interest(
    principal: float,
    rate: float,
    years: int
) -> str:
    '''
    Calculate compound interest.
    Use this tool when the user asks about investment growth,
    compound interest, or future value of money.
    Args:
        principal: Initial investment amount in dollars
        rate: Annual interest rate as a decimal (e.g., 0.08 for 8%)
        years: Number of years to compound
    '''
    amount = principal * (1 + rate) ** years
    interest = amount - principal
    return f'Future value: ${amount:,.2f} | Interest earned: ${interest:,.2f}'
# Test the tool directly
print(calculate_compound_interest.invoke({
    'principal': 10000, 'rate': 0.08, 'years': 10
}))

from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain.tools import tool
# Web search tool
search_tool = DuckDuckGoSearchRun()
# Wikipedia tool
wiki_tool = WikipediaQueryRun(
    api_wrapper=WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=500)
)
# Test built-in tools
print(search_tool.invoke('LangChain latest version 2026'))
print(wiki_tool.invoke('Large Language Model'))

# Inspect a tool's schema
print(calculate_compound_interest.name)
# 'calculate_compound_interest'
print(calculate_compound_interest.description)
# 'Calculate compound interest. Use this tool when...'
print(calculate_compound_interest.args_schema.schema())
# {'properties': {'principal': {'type': 'number'}, 'rate': ..., 'years': ...}}

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# Store for conversation histories, keyed by session ID
store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]
# Wrap your chain with message history
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key  = 'input',
    history_messages_key= 'chat_history',
)
# Invoke with a session ID
config = {'configurable': {'session_id': 'user_abc'}}
response1 = chain_with_history.invoke({'input': 'My name is Alice.'}, config=config)
response2 = chain_with_history.invoke({'input': 'What is my name?'},   config=config)
print(response2)  # Agent remembers: 'Your name is Alice.'

# pip install langchain-redis
from langchain_redis import RedisChatMessageHistory
def get_redis_history(session_id: str):
    return RedisChatMessageHistory(
        session_id = session_id,
        url        = 'redis://localhost:6379'
    )
chain_with_redis = RunnableWithMessageHistory(
    chain,
    get_redis_history,
    input_messages_key  = 'input',
    history_messages_key= 'chat_history',
)
# Now conversation history survives server restarts

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_core.tools import tool
# LLM
llm = ChatOpenAI(model='gpt-4o', temperature=0)
# Built-in tools
search = DuckDuckGoSearchRun()
wiki   = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=1))
# Custom tool
@tool
def calculate(expression: str) -> str:
    '''Evaluate a mathematical expression. Use for any arithmetic or calculation.
    Input must be a valid Python math expression, e.g. '2 ** 10' or '100 * 0.15'.'''
    try:
        return str(eval(expression, {'__builtins__': {}}, {}))
    except Exception as e:
        return f'Error: {e}'
# Collect all tools
tools = [search, wiki, calculate]

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
# Pull the standard ReAct prompt from LangChain Hub
# This prompt instructs the LLM to follow Thought/Action/Observation format
react_prompt = hub.pull('hwchase17/react')
# Create the agent (LLM + tools + prompt wired together)
agent = create_react_agent(
    llm    = llm,
    tools  = tools,
    prompt = react_prompt
)
# Wrap in AgentExecutor — this is the runtime loop
agent_executor = AgentExecutor(
    agent           = agent,
    tools           = tools,
    verbose         = True,    # print each Thought/Action/Observation
    max_iterations  = 10,      # prevent infinite loops
    handle_parsing_errors = True
)

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# Prompt with placeholders for chat history and agent scratchpad
prompt = ChatPromptTemplate.from_messages([
    ('system', 'You are a helpful AI assistant with access to tools. '
               'Use tools whenever they would help answer the question accurately.'),
    MessagesPlaceholder(variable_name='chat_history', optional=True),
    ('human', '{input}'),
    MessagesPlaceholder(variable_name='agent_scratchpad'),
])
# Create the tool-calling agent
agent = create_tool_calling_agent(
    llm    = llm,
    tools  = tools,
    prompt = prompt
)
agent_executor = AgentExecutor(
    agent          = agent,
    tools          = tools,
    verbose        = True,
    max_iterations = 10
)
# Invoke
result = agent_executor.invoke({'input': 'Search for LangChain 2026 updates'})
print(result['output'])

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# Session store
session_store = {}
def get_history(session_id: str):
    if session_id not in session_store:
        session_store[session_id] = InMemoryChatMessageHistory()
    return session_store[session_id]
# Wrap the executor with message history
agent_with_memory = RunnableWithMessageHistory(
    agent_executor,
    get_history,
    input_messages_key   = 'input',
    history_messages_key = 'chat_history',
)
config = {'configurable': {'session_id': 'session_001'}}
# Turn 1
r1 = agent_with_memory.invoke(
    {'input': 'My name is Alice and I am interested in investing $50,000.'},
    config=config
)
print(r1['output'])
# Turn 2 — agent remembers context from Turn 1
r2 = agent_with_memory.invoke(
    {'input': 'What would my investment be worth in 10 years at 8% interest?'},
    config=config
)
print(r2['output'])
# Agent uses calculate tool with principal=50000, rate=0.08, years=10
# And addresses the user as Alice

SYSTEM_PROMPT = '''You are FinBot, an expert financial analysis assistant.
Your capabilities:
- Search for current financial news and market data
- Perform precise financial calculations
- Explain complex financial concepts in simple terms
Your rules:
- Always use tools to verify current data before answering
- Never fabricate financial figures — if unsure, search first
- Present numbers in a clear, formatted way
- Always include a disclaimer that this is not financial advice
Respond concisely and professionally.'''
prompt = ChatPromptTemplate.from_messages([
    ('system', SYSTEM_PROMPT),
    MessagesPlaceholder(variable_name='chat_history', optional=True),
    ('human', '{input}'),
    MessagesPlaceholder(variable_name='agent_scratchpad'),
])
finbot = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
finbot_executor = AgentExecutor(agent=finbot, tools=tools, verbose=True)

# 1. Handle tool errors gracefully
@tool
def safe_search(query: str) -> str:
    '''Search the web for current information.'''
    try:
        result = DuckDuckGoSearchRun().run(query)
        return result if result else 'No results found for this query.'
    except Exception as e:
        return f'Search failed: {str(e)}. Try rephrasing the query.'
# 2. Limit iterations and handle errors in AgentExecutor
agent_executor = AgentExecutor(
    agent                 = agent,
    tools                 = tools,
    max_iterations        = 8,          # stop after 8 steps
    max_execution_time    = 60,          # stop after 60 seconds
    handle_parsing_errors = True,        # recover from LLM output parse errors
    early_stopping_method = 'generate',  # generate a final answer if limit hit
)
# 3. Catch executor-level errors
try:
    result = agent_executor.invoke({'input': user_query})
except Exception as e:
    result = {'output': f'Agent encountered an error: {str(e)}'}

import os
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.agents import create_tool_calling_agent, AgentExecutor
os.environ['OPENAI_API_KEY'] = 'your-api-key'
# ── LLM ──
llm = ChatOpenAI(model='gpt-4o', temperature=0)
# ── TOOLS ──
search = DuckDuckGoSearchRun()
wiki   = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=1))
@tool
def calculate(expression: str) -> str:
    '''Evaluate a Python math expression. Input: valid Python expression string.'''
    try:
        return str(eval(expression, {'__builtins__': {}}, {}))
    except Exception as e:
        return f'Calculation error: {e}'
tools = [search, wiki, calculate]
# ── PROMPT ──
prompt = ChatPromptTemplate.from_messages([
    ('system', 'You are a helpful AI assistant. Use tools when needed.'),
    MessagesPlaceholder('chat_history', optional=True),
    ('human', '{input}'),
    MessagesPlaceholder('agent_scratchpad'),
])
# ── AGENT ──
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(
    agent=agent, tools=tools, verbose=True,
    max_iterations=10, handle_parsing_errors=True
)
# ── MEMORY ──
store = {}
def get_history(sid): 
    store.setdefault(sid, InMemoryChatMessageHistory())
    return store[sid]
agent = RunnableWithMessageHistory(
    executor, get_history,
    input_messages_key='input',
    history_messages_key='chat_history'
)
# ── RUN ──
cfg = {'configurable': {'session_id': 'demo'}}
while True:
    user_input = input('You: ')
    if user_input.lower() in ['quit', 'exit']: break
    response = agent.invoke({'input': user_input}, config=cfg)
    print(f'Agent: {response["output"]}')