Building Your First AI Agent with LangChain and Python

Published on December 15, 2024 • 14 min read

LangChain has revolutionized AI agent development by providing a powerful framework for building intelligent applications. This comprehensive guide teaches you how to create sophisticated AI agents that can reason, use tools, and interact with external systems.

Table of Contents

1. Introduction to LangChain
2. Setting Up Your Environment
3. Basic Agent Architecture
4. Building Your First Agent
5. Adding Tools and Capabilities
6. Memory and State Management
7. Advanced Agent Patterns
8. Production Deployment

Introduction to LangChain {#introduction}

LangChain is a framework for developing applications powered by language models. It enables you to build agents that can:

• Reason: Make decisions based on context and goals
• Use Tools: Interact with APIs, databases, and external services
• Remember: Maintain conversation history and context
• Plan: Break down complex tasks into steps

Key Components

• Agents: The core reasoning engine
• Tools: Functions the agent can call
• Memory: Storage for conversation history
• Chains: Sequences of operations
• Prompts: Templates for model interactions

Setting Up Your Environment {#setup}

Installation and Dependencies

1pip install langchain openai python-dotenv
2pip install langchain-community langchain-experimental
3pip install faiss-cpu  # For vector storage
4pip install requests beautifulsoup4  # For web scraping tools

Environment Configuration

1import os
2from dotenv import load_dotenv
3from langchain.llms import OpenAI
4from langchain.chat_models import ChatOpenAI
5from langchain.agents import initialize_agent, AgentType
6from langchain.tools import Tool
7from langchain.memory import ConversationBufferMemory
8
9# Load environment variables
10load_dotenv()
11
12# Configure OpenAI
13os.environ["OPENAI_API_KEY"] = "your_openai_api_key"
14
15# Initialize language model
16llm = ChatOpenAI(
17    model_name="gpt-3.5-turbo",
18    temperature=0.7,
19    max_tokens=1000
20)
21
22print("LangChain environment configured successfully!")

Basic Agent Architecture {#architecture}

Understanding Agent Components

1from langchain.agents import AgentExecutor
2from langchain.agents.format_scratchpad import format_to_openai_function_messages
3from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
4from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
5from langchain.tools.render import format_tool_to_openai_function
6
7class BaseAgent:
8    def __init__(self, llm, tools, memory=None):
9        self.llm = llm
10        self.tools = tools
11        self.memory = memory or ConversationBufferMemory(
12            memory_key="chat_history",
13            return_messages=True
14        )
15        self.agent_executor = self._create_agent()
16    
17    def _create_agent(self):
18        """Create the agent executor"""
19        # Create prompt template
20        prompt = ChatPromptTemplate.from_messages([
21            ("system", """You are a helpful AI assistant. Use the available tools to help users accomplish their tasks.
22            
23            Available tools:
24            {tools}
25            
26            Always think step by step and explain your reasoning.
27            """),
28            MessagesPlaceholder(variable_name="chat_history"),
29            ("user", "{input}"),
30            MessagesPlaceholder(variable_name="agent_scratchpad")
31        ])
32        
33        # Format tools for OpenAI functions
34        llm_with_tools = self.llm.bind(
35            functions=[format_tool_to_openai_function(t) for t in self.tools]
36        )
37        
38        # Create agent
39        agent = (
40            {
41                "input": lambda x: x["input"],
42                "agent_scratchpad": lambda x: format_to_openai_function_messages(
43                    x["intermediate_steps"]
44                ),
45                "chat_history": lambda x: x["chat_history"],
46                "tools": lambda x: "\n".join([f"{tool.name}: {tool.description}" for tool in self.tools])
47            }
48            | prompt
49            | llm_with_tools
50            | OpenAIFunctionsAgentOutputParser()
51        )
52        
53        # Create executor
54        agent_executor = AgentExecutor(
55            agent=agent,
56            tools=self.tools,
57            memory=self.memory,
58            verbose=True,
59            max_iterations=5
60        )
61        
62        return agent_executor
63    
64    def run(self, input_text):
65        """Run the agent with input"""
66        return self.agent_executor.invoke({"input": input_text})
67
68# Usage example
69tools = []  # We'll add tools in the next section
70agent = BaseAgent(llm, tools)

Building Your First Agent {#first-agent}

Creating Basic Tools

1import requests
2from datetime import datetime
3import json
4
5def get_current_time():
6    """Get the current time"""
7    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
8
9def calculate_math(expression):
10    """Calculate mathematical expressions safely"""
11    try:
12        # Only allow safe mathematical operations
13        allowed_chars = set('0123456789+-*/.() ')
14        if not all(c in allowed_chars for c in expression):
15            return "Error: Invalid characters in expression"
16        
17        result = eval(expression)
18        return f"Result: {result}"
19    except Exception as e:
20        return f"Error: {str(e)}"
21
22def search_web(query):
23    """Search the web for information"""
24    try:
25        # Using a simple search API (replace with your preferred service)
26        url = f"https://api.duckduckgo.com/?q={query}&format=json&no_html=1&skip_disambig=1"
27        response = requests.get(url)
28        data = response.json()
29        
30        if data.get('Abstract'):
31            return data['Abstract']
32        elif data.get('RelatedTopics'):
33            return data['RelatedTopics'][0].get('Text', 'No results found')
34        else:
35            return "No results found"
36    except Exception as e:
37        return f"Search error: {str(e)}"
38
39def write_file(filename, content):
40    """Write content to a file"""
41    try:
42        with open(filename, 'w') as f:
43            f.write(content)
44        return f"Successfully wrote to {filename}"
45    except Exception as e:
46        return f"Error writing file: {str(e)}"
47
48def read_file(filename):
49    """Read content from a file"""
50    try:
51        with open(filename, 'r') as f:
52            content = f.read()
53        return content
54    except Exception as e:
55        return f"Error reading file: {str(e)}"
56
57# Create LangChain tools
58tools = [
59    Tool(
60        name="get_current_time",
61        description="Get the current date and time",
62        func=get_current_time
63    ),
64    Tool(
65        name="calculate_math",
66        description="Calculate mathematical expressions. Input should be a valid mathematical expression.",
67        func=calculate_math
68    ),
69    Tool(
70        name="search_web",
71        description="Search the web for information. Input should be a search query.",
72        func=search_web
73    ),
74    Tool(
75        name="write_file",
76        description="Write content to a file. Input should be 'filename|content'.",
77        func=lambda x: write_file(*x.split('|', 1))
78    ),
79    Tool(
80        name="read_file",
81        description="Read content from a file. Input should be the filename.",
82        func=read_file
83    )
84]
85
86# Create agent with tools
87agent = BaseAgent(llm, tools)
88
89# Test the agent
90response = agent.run("What time is it and what's 25 * 4?")
91print(response)

Simple Agent Example

1class SimpleTaskAgent:
2    def __init__(self):
3        self.llm = ChatOpenAI(temperature=0.7)
4        self.tools = self._create_tools()
5        self.memory = ConversationBufferMemory(
6            memory_key="chat_history",
7            return_messages=True
8        )
9        
10        self.agent = initialize_agent(
11            tools=self.tools,
12            llm=self.llm,
13            agent=AgentType.OPENAI_FUNCTIONS,
14            memory=self.memory,
15            verbose=True
16        )
17    
18    def _create_tools(self):
19        """Create basic tools for the agent"""
20        return [
21            Tool(
22                name="calculator",
23                description="Useful for mathematical calculations",
24                func=calculate_math
25            ),
26            Tool(
27                name="timer",
28                description="Get current time and date",
29                func=get_current_time
30            ),
31            Tool(
32                name="web_search",
33                description="Search the internet for information",
34                func=search_web
35            )
36        ]
37    
38    def chat(self, message):
39        """Chat with the agent"""
40        try:
41            response = self.agent.run(message)
42            return response
43        except Exception as e:
44            return f"Error: {str(e)}"
45
46# Usage
47simple_agent = SimpleTaskAgent()
48print(simple_agent.chat("What's the weather like today in New York?"))
49print(simple_agent.chat("Calculate 15% tip on a $45 bill"))

Adding Tools and Capabilities {#tools}

Database Integration Tool

1import sqlite3
2from langchain.tools import BaseTool
3from typing import Optional, Type
4from pydantic import BaseModel, Field
5
6class DatabaseQueryInput(BaseModel):
7    query: str = Field(description="SQL query to execute")
8
9class DatabaseTool(BaseTool):
10    name = "database_query"
11    description = "Execute SQL queries on the database"
12    args_schema: Type[BaseModel] = DatabaseQueryInput
13    
14    def __init__(self, db_path: str):
15        super().__init__()
16        self.db_path = db_path
17        self._init_database()
18    
19    def _init_database(self):
20        """Initialize sample database"""
21        conn = sqlite3.connect(self.db_path)
22        cursor = conn.cursor()
23        
24        # Create sample table
25        cursor.execute('''
26            CREATE TABLE IF NOT EXISTS users (
27                id INTEGER PRIMARY KEY,
28                name TEXT NOT NULL,
29                email TEXT NOT NULL,
30                age INTEGER
31            )
32        ''')
33        
34        # Insert sample data
35        cursor.execute("INSERT OR IGNORE INTO users (id, name, email, age) VALUES (1, 'John Doe', 'john@example.com', 30)")
36        cursor.execute("INSERT OR IGNORE INTO users (id, name, email, age) VALUES (2, 'Jane Smith', 'jane@example.com', 25)")
37        
38        conn.commit()
39        conn.close()
40    
41    def _run(self, query: str) -> str:
42        """Execute database query"""
43        try:
44            conn = sqlite3.connect(self.db_path)
45            cursor = conn.cursor()
46            
47            cursor.execute(query)
48            
49            if query.strip().upper().startswith('SELECT'):
50                results = cursor.fetchall()
51                columns = [description[0] for description in cursor.description]
52                
53                if results:
54                    formatted_results = []
55                    for row in results:
56                        row_dict = dict(zip(columns, row))
57                        formatted_results.append(row_dict)
58                    return json.dumps(formatted_results, indent=2)
59                else:
60                    return "No results found"
61            else:
62                conn.commit()
63                return f"Query executed successfully. Rows affected: {cursor.rowcount}"
64                
65        except Exception as e:
66            return f"Database error: {str(e)}"
67        finally:
68            conn.close()
69
70# API Integration Tool
71class APITool(BaseTool):
72    name = "api_request"
73    description = "Make HTTP requests to APIs"
74    
75    def _run(self, url: str, method: str = "GET", data: dict = None) -> str:
76        """Make API request"""
77        try:
78            if method.upper() == "GET":
79                response = requests.get(url)
80            elif method.upper() == "POST":
81                response = requests.post(url, json=data)
82            else:
83                return f"Unsupported method: {method}"
84            
85            return f"Status: {response.status_code}\nResponse: {response.text[:500]}"
86        except Exception as e:
87            return f"API error: {str(e)}"
88
89# Email Tool
90class EmailTool(BaseTool):
91    name = "send_email"
92    description = "Send emails (simulation)"
93    
94    def _run(self, to: str, subject: str, body: str) -> str:
95        """Send email (simulated)"""
96        # In production, integrate with actual email service
97        email_log = {
98            "to": to,
99            "subject": subject,
100            "body": body,
101            "timestamp": datetime.now().isoformat()
102        }
103        
104        # Log email instead of actually sending
105        with open("email_log.json", "a") as f:
106            f.write(json.dumps(email_log) + "\n")
107        
108        return f"Email sent to {to} with subject '{subject}'"
109
110# Create enhanced agent with new tools
111enhanced_tools = tools + [
112    DatabaseTool("sample.db"),
113    APITool(),
114    EmailTool()
115]
116
117enhanced_agent = BaseAgent(llm, enhanced_tools)

Custom Tool Creation

1from langchain.tools import StructuredTool
2from pydantic import BaseModel, Field
3
4class WeatherInput(BaseModel):
5    location: str = Field(description="City name or location")
6    units: str = Field(default="metric", description="Temperature units (metric/imperial)")
7
8def get_weather(location: str, units: str = "metric") -> str:
9    """Get weather information for a location"""
10    # Simulate weather API call
11    weather_data = {
12        "location": location,
13        "temperature": "22°C" if units == "metric" else "72°F",
14        "condition": "Partly cloudy",
15        "humidity": "65%",
16        "wind": "10 km/h" if units == "metric" else "6 mph"
17    }
18    
19    return f"Weather in {location}: {weather_data['temperature']}, {weather_data['condition']}, Humidity: {weather_data['humidity']}, Wind: {weather_data['wind']}"
20
21# Create structured tool
22weather_tool = StructuredTool.from_function(
23    func=get_weather,
24    name="get_weather",
25    description="Get current weather information for a specific location",
26    args_schema=WeatherInput
27)
28
29# Add to tools list
30all_tools = enhanced_tools + [weather_tool]

Memory and State Management {#memory}

Different Memory Types

1from langchain.memory import (
2    ConversationBufferMemory,
3    ConversationBufferWindowMemory,
4    ConversationSummaryMemory,
5    ConversationSummaryBufferMemory
6)
7
8class MemoryManager:
9    def __init__(self, llm):
10        self.llm = llm
11        
12    def create_buffer_memory(self, return_messages=True):
13        """Create simple buffer memory"""
14        return ConversationBufferMemory(
15            memory_key="chat_history",
16            return_messages=return_messages
17        )
18    
19    def create_window_memory(self, k=5):
20        """Create window memory (keeps last k exchanges)"""
21        return ConversationBufferWindowMemory(
22            memory_key="chat_history",
23            k=k,
24            return_messages=True
25        )
26    
27    def create_summary_memory(self):
28        """Create summary memory (summarizes old conversations)"""
29        return ConversationSummaryMemory(
30            llm=self.llm,
31            memory_key="chat_history",
32            return_messages=True
33        )
34    
35    def create_summary_buffer_memory(self, max_token_limit=1000):
36        """Create summary buffer memory (hybrid approach)"""
37        return ConversationSummaryBufferMemory(
38            llm=self.llm,
39            memory_key="chat_history",
40            max_token_limit=max_token_limit,
41            return_messages=True
42        )
43
44# Usage
45memory_manager = MemoryManager(llm)
46smart_memory = memory_manager.create_summary_buffer_memory()
47
48# Create agent with smart memory
49smart_agent = BaseAgent(llm, all_tools, smart_memory)

Persistent Memory with Vector Storage

1from langchain.vectorstores import FAISS
2from langchain.embeddings import OpenAIEmbeddings
3from langchain.memory import VectorStoreRetrieverMemory
4
5class PersistentMemoryAgent:
6    def __init__(self, llm, tools):
7        self.llm = llm
8        self.tools = tools
9        self.embeddings = OpenAIEmbeddings()
10        self.vector_store = self._create_vector_store()
11        self.memory = self._create_vector_memory()
12        self.agent = self._create_agent()
13    
14    def _create_vector_store(self):
15        """Create vector store for memory"""
16        # Initialize with empty documents
17        texts = ["Initial conversation started"]
18        return FAISS.from_texts(texts, self.embeddings)
19    
20    def _create_vector_memory(self):
21        """Create vector-based memory"""
22        retriever = self.vector_store.as_retriever(search_kwargs={"k": 5})
23        return VectorStoreRetrieverMemory(
24            retriever=retriever,
25            memory_key="chat_history"
26        )
27    
28    def _create_agent(self):
29        """Create agent with vector memory"""
30        return initialize_agent(
31            tools=self.tools,
32            llm=self.llm,
33            agent=AgentType.OPENAI_FUNCTIONS,
34            memory=self.memory,
35            verbose=True
36        )
37    
38    def chat(self, message):
39        """Chat with persistent memory"""
40        response = self.agent.run(message)
41        
42        # Add to vector store for future retrieval
43        self.vector_store.add_texts([f"User: {message}\nAssistant: {response}"])
44        
45        return response
46    
47    def save_memory(self, path):
48        """Save vector store to disk"""
49        self.vector_store.save_local(path)
50    
51    def load_memory(self, path):
52        """Load vector store from disk"""
53        self.vector_store = FAISS.load_local(path, self.embeddings)
54        self.memory = self._create_vector_memory()
55        self.agent = self._create_agent()
56
57# Usage
58persistent_agent = PersistentMemoryAgent(llm, all_tools)

Advanced Agent Patterns {#advanced}

Multi-Agent System

1class SpecializedAgent:
2    def __init__(self, name, role, llm, tools, system_prompt):
3        self.name = name
4        self.role = role
5        self.llm = llm
6        self.tools = tools
7        self.system_prompt = system_prompt
8        self.memory = ConversationBufferMemory(
9            memory_key="chat_history",
10            return_messages=True
11        )
12        self.agent = self._create_agent()
13    
14    def _create_agent(self):
15        """Create specialized agent"""
16        prompt = ChatPromptTemplate.from_messages([
17            ("system", self.system_prompt),
18            MessagesPlaceholder(variable_name="chat_history"),
19            ("user", "{input}"),
20            MessagesPlaceholder(variable_name="agent_scratchpad")
21        ])
22        
23        llm_with_tools = self.llm.bind(
24            functions=[format_tool_to_openai_function(t) for t in self.tools]
25        )
26        
27        agent = (
28            {
29                "input": lambda x: x["input"],
30                "agent_scratchpad": lambda x: format_to_openai_function_messages(
31                    x["intermediate_steps"]
32                ),
33                "chat_history": lambda x: x["chat_history"]
34            }
35            | prompt
36            | llm_with_tools
37            | OpenAIFunctionsAgentOutputParser()
38        )
39        
40        return AgentExecutor(
41            agent=agent,
42            tools=self.tools,
43            memory=self.memory,
44            verbose=True
45        )
46    
47    def run(self, input_text):
48        """Run the specialized agent"""
49        return self.agent.invoke({"input": input_text})
50
51class MultiAgentSystem:
52    def __init__(self, llm):
53        self.llm = llm
54        self.agents = self._create_agents()
55        self.coordinator = self._create_coordinator()
56    
57    def _create_agents(self):
58        """Create specialized agents"""
59        agents = {}
60        
61        # Research Agent
62        agents['researcher'] = SpecializedAgent(
63            name="researcher",
64            role="Research and information gathering",
65            llm=self.llm,
66            tools=[weather_tool, Tool(name="web_search", description="Search web", func=search_web)],
67            system_prompt="You are a research specialist. Your job is to gather accurate information from various sources."
68        )
69        
70        # Data Agent
71        agents['data_analyst'] = SpecializedAgent(
72            name="data_analyst",
73            role="Data analysis and calculations",
74            llm=self.llm,
75            tools=[Tool(name="calculator", description="Calculate math", func=calculate_math), DatabaseTool("sample.db")],
76            system_prompt="You are a data analysis expert. You excel at mathematical calculations and data interpretation."
77        )
78        
79        # Communication Agent
80        agents['communicator'] = SpecializedAgent(
81            name="communicator",
82            role="Communication and file operations",
83            llm=self.llm,
84            tools=[EmailTool(), Tool(name="write_file", description="Write files", func=lambda x: write_file(*x.split('|', 1)))],
85            system_prompt="You are a communication specialist. You handle all external communications and file operations."
86        )
87        
88        return agents
89    
90    def _create_coordinator(self):
91        """Create coordinator agent"""
92        return SpecializedAgent(
93            name="coordinator",
94            role="Task coordination and delegation",
95            llm=self.llm,
96            tools=[],
97            system_prompt="""You are a coordinator agent. Your job is to:
98            1. Analyze user requests
99            2. Determine which specialized agents should handle different parts
100            3. Coordinate the workflow between agents
101            4. Provide final responses to users
102            
103            Available agents:
104            - researcher: For gathering information and research
105            - data_analyst: For calculations and data analysis
106            - communicator: For sending emails and file operations
107            """
108        )
109    
110    def process_request(self, user_input):
111        """Process user request through multi-agent system"""
112        # Coordinator analyzes the request
113        coordination_prompt = f"""
114        User request: {user_input}
115        
116        Analyze this request and determine:
117        1. Which agents should be involved
118        2. What tasks each agent should perform
119        3. The order of operations
120        
121        Provide a plan in JSON format with agent assignments.
122        """
123        
124        plan = self.coordinator.run(coordination_prompt)
125        
126        # Execute plan (simplified version)
127        results = {}
128        
129        # For demonstration, route based on keywords
130        if any(word in user_input.lower() for word in ['weather', 'search', 'find', 'research']):
131            results['research'] = self.agents['researcher'].run(user_input)
132        
133        if any(word in user_input.lower() for word in ['calculate', 'math', 'data', 'analyze']):
134            results['analysis'] = self.agents['data_analyst'].run(user_input)
135        
136        if any(word in user_input.lower() for word in ['email', 'send', 'write', 'file']):
137            results['communication'] = self.agents['communicator'].run(user_input)
138        
139        # Coordinator synthesizes final response
140        synthesis_prompt = f"""
141        User request: {user_input}
142        Agent results: {json.dumps(results, indent=2)}
143        
144        Provide a comprehensive response to the user based on all agent results.
145        """
146        
147        final_response = self.coordinator.run(synthesis_prompt)
148        return final_response
149
150# Usage
151multi_agent_system = MultiAgentSystem(llm)
152response = multi_agent_system.process_request("What's the weather in Paris and calculate 15% of 200?")
153print(response)

Self-Improving Agent

1class SelfImprovingAgent:
2    def __init__(self, llm, tools):
3        self.llm = llm
4        self.tools = tools
5        self.performance_log = []
6        self.improvement_suggestions = []
7        
8    def run_with_feedback(self, user_input):
9        """Run agent and collect performance feedback"""
10        start_time = datetime.now()
11        
12        try:
13            # Create agent
14            agent = initialize_agent(
15                tools=self.tools,
16                llm=self.llm,
17                agent=AgentType.OPENAI_FUNCTIONS,
18                verbose=True
19            )
20            
21            # Execute task
22            response = agent.run(user_input)
23            
24            # Log performance
25            execution_time = (datetime.now() - start_time).total_seconds()
26            
27            self.performance_log.append({
28                'input': user_input,
29                'response': response,
30                'execution_time': execution_time,
31                'timestamp': datetime.now().isoformat(),
32                'success': True
33            })
34            
35            # Analyze performance
36            self._analyze_performance()
37            
38            return response
39            
40        except Exception as e:
41            # Log failure
42            self.performance_log.append({
43                'input': user_input,
44                'error': str(e),
45                'execution_time': (datetime.now() - start_time).total_seconds(),
46                'timestamp': datetime.now().isoformat(),
47                'success': False
48            })
49            
50            return f"Error: {str(e)}"
51    
52    def _analyze_performance(self):
53        """Analyze recent performance and suggest improvements"""
54        if len(self.performance_log) < 5:
55            return
56        
57        recent_logs = self.performance_log[-5:]
58        
59        # Calculate metrics
60        success_rate = sum(1 for log in recent_logs if log['success']) / len(recent_logs)
61        avg_execution_time = sum(log['execution_time'] for log in recent_logs) / len(recent_logs)
62        
63        # Generate improvement suggestions
64        if success_rate < 0.8:
65            self.improvement_suggestions.append({
66                'type': 'error_handling',
67                'suggestion': 'Improve error handling and input validation',
68                'timestamp': datetime.now().isoformat()
69            })
70        
71        if avg_execution_time > 10:
72            self.improvement_suggestions.append({
73                'type': 'performance',
74                'suggestion': 'Optimize tool usage and reduce execution time',
75                'timestamp': datetime.now().isoformat()
76            })
77    
78    def get_performance_report(self):
79        """Generate performance report"""
80        if not self.performance_log:
81            return "No performance data available"
82        
83        total_runs = len(self.performance_log)
84        successful_runs = sum(1 for log in self.performance_log if log['success'])
85        success_rate = successful_runs / total_runs
86        avg_time = sum(log['execution_time'] for log in self.performance_log) / total_runs
87        
88        report = f"""
89        Performance Report:
90        - Total runs: {total_runs}
91        - Success rate: {success_rate:.2%}
92        - Average execution time: {avg_time:.2f} seconds
93        - Recent improvements suggested: {len(self.improvement_suggestions)}
94        """
95        
96        return report
97
98# Usage
99self_improving_agent = SelfImprovingAgent(llm, all_tools)

Production Deployment {#deployment}

FastAPI Web Service

1from fastapi import FastAPI, HTTPException
2from pydantic import BaseModel
3import uvicorn
4
5app = FastAPI(title="LangChain Agent API")
6
7# Initialize agent
8production_agent = BaseAgent(llm, all_tools)
9
10class ChatRequest(BaseModel):
11    message: str
12    session_id: str = "default"
13
14class ChatResponse(BaseModel):
15    response: str
16    session_id: str
17
18@app.post("/chat", response_model=ChatResponse)
19async def chat_endpoint(request: ChatRequest):
20    """Chat with the AI agent"""
21    try:
22        response = production_agent.run(request.message)
23        return ChatResponse(
24            response=response['output'],
25            session_id=request.session_id
26        )
27    except Exception as e:
28        raise HTTPException(status_code=500, detail=str(e))
29
30@app.get("/health")
31async def health_check():
32    """Health check endpoint"""
33    return {"status": "healthy", "agent": "ready"}
34
35@app.get("/tools")
36async def list_tools():
37    """List available tools"""
38    return {
39        "tools": [
40            {"name": tool.name, "description": tool.description}
41            for tool in all_tools
42        ]
43    }
44
45if __name__ == "__main__":
46    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker Deployment

1FROM python:3.9-slim
2
3WORKDIR /app
4
5# Install dependencies
6COPY requirements.txt .
7RUN pip install --no-cache-dir -r requirements.txt
8
9# Copy application
10COPY . .
11
12# Expose port
13EXPOSE 8000
14
15# Run application
16CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Monitoring and Logging

1import logging
2from datetime import datetime
3import json
4
5class AgentMonitor:
6    def __init__(self, agent):
7        self.agent = agent
8        self.logger = self._setup_logging()
9        
10    def _setup_logging(self):
11        """Setup logging configuration"""
12        logging.basicConfig(
13            level=logging.INFO,
14            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
15            handlers=[
16                logging.FileHandler('agent.log'),
17                logging.StreamHandler()
18            ]
19        )
20        return logging.getLogger('LangChainAgent')
21    
22    def monitored_run(self, user_input, session_id="default"):
23        """Run agent with monitoring"""
24        start_time = datetime.now()
25        
26        self.logger.info(f"Session {session_id}: Processing request: {user_input}")
27        
28        try:
29            response = self.agent.run(user_input)
30            execution_time = (datetime.now() - start_time).total_seconds()
31            
32            self.logger.info(f"Session {session_id}: Request completed in {execution_time:.2f}s")
33            
34            # Log metrics
35            metrics = {
36                'session_id': session_id,
37                'input_length': len(user_input),
38                'response_length': len(response['output']),
39                'execution_time': execution_time,
40                'timestamp': start_time.isoformat(),
41                'success': True
42            }
43            
44            self.logger.info(f"Metrics: {json.dumps(metrics)}")
45            
46            return response
47            
48        except Exception as e:
49            execution_time = (datetime.now() - start_time).total_seconds()
50            
51            self.logger.error(f"Session {session_id}: Error after {execution_time:.2f}s: {str(e)}")
52            
53            # Log error metrics
54            error_metrics = {
55                'session_id': session_id,
56                'input_length': len(user_input),
57                'execution_time': execution_time,
58                'timestamp': start_time.isoformat(),
59                'success': False,
60                'error': str(e)
61            }
62            
63            self.logger.error(f"Error metrics: {json.dumps(error_metrics)}")
64            
65            raise
66
67# Usage
68monitored_agent = AgentMonitor(production_agent)

Best Practices

Error Handling and Resilience

1class ResilientAgent:
2    def __init__(self, llm, tools, max_retries=3):
3        self.llm = llm
4        self.tools = tools
5        self.max_retries = max_retries
6        self.agent = self._create_agent()
7    
8    def _create_agent(self):
9        """Create agent with error handling"""
10        return initialize_agent(
11            tools=self.tools,
12            llm=self.llm,
13            agent=AgentType.OPENAI_FUNCTIONS,
14            handle_parsing_errors=True,
15            max_iterations=5,
16            early_stopping_method="generate"
17        )
18    
19    def run_with_retry(self, user_input):
20        """Run agent with retry logic"""
21        for attempt in range(self.max_retries):
22            try:
23                response = self.agent.run(user_input)
24                return response
25            except Exception as e:
26                if attempt == self.max_retries - 1:
27                    return f"Sorry, I encountered an error after {self.max_retries} attempts: {str(e)}"
28                
29                print(f"Attempt {attempt + 1} failed: {str(e)}. Retrying...")
30                time.sleep(1)  # Brief delay before retry
31
32# Usage
33resilient_agent = ResilientAgent(llm, all_tools)

Conclusion

LangChain provides a powerful framework for building sophisticated AI agents. Start with simple tools and gradually add complexity as you understand the patterns and requirements of your specific use case.

Key takeaways:

• Start simple and iterate
• Design tools carefully for your domain
• Implement proper error handling and monitoring
• Use appropriate memory strategies for your use case
• Consider multi-agent architectures for complex tasks

---

This guide provides a comprehensive foundation for building AI agents with LangChain. Experiment with different configurations to find what works best for your specific requirements.