LangChain Agent开发实战：从环境安装到生产级对话系统的完整代码

一、背景：为什么需要Agent

在AI应用中，单纯的问答（QA）模式有两大局限：

1. 无法执行操作：只能回答问题，不能替用户执行任务（如查天气、搜信息、发邮件）
2. 知识有截止日期：LLM无法获取实时信息，无法访问私有数据

Agent（智能体）正是为解决这两个问题而生的。Agent的核心能力是：让LLM拥有"思考 → 决策 → 调用工具 → 观察结果 → 继续思考"的循环能力，从而能够自主完成复杂任务。

本文手把手实现一个基于LangChain的生产级ReAct Agent，支持：天气查询、网页搜索、文件操作、数学计算、日程管理。

二、方案：LangChain Agent完整开发步骤

2.1 Python环境安装

# 推荐使用conda创建独立环境，避免依赖冲突
# Step 1: 安装Anaconda（如果还没有）
wget https://repo.anaconda.com/archive/Anaconda3-2024.02-1-Linux-x86_64.sh
bash Anaconda3-2024.02-1-Linux-x86_64.sh -b -p ~/anaconda3
echo 'export PATH=~/anaconda3/bin:$PATH' >> ~/.bashrc
source ~/.bashrc

# Step 2: 创建LangChain专用环境
conda create -n langchain python=3.11 -y
conda activate langchain

# Step 3: 安装核心依赖
# LangChain + OpenAI
pip install langchain langchain-openai openai

# LangChain社区包（包含很多工具）
pip install langchain-community

# Agent核心组件
pip install langchain-core langchain-experimental

# 工具生态
pip install requests     # HTTP请求
pip install duckduckgo-search  # 搜索
pip install wikipedia    # 维基百科
pip install python-dateutil  # 日期处理
pip install numexpr     # 数学计算（安全）

# 向量数据库（可选，用于知识库）
pip install chromadb faiss-cpu

# Web服务
pip install fastapi uvicorn

# 日志和监控
pip install langsmith  # LangChain官方监控平台

# 验证安装
python -c "import langchain; print(langchain.__version__)"
# 正常输出：0.2.x 或更高

2.2 第一个简单Agent

"""
LangChain ReAct Agent 入门示例
演示Agent的基本工作原理
"""

from langchain.agents import AgentType, initialize_agent
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI

# 初始化LLM（使用OpenAI GPT-4）
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    api_key="sk-your-api-key-here"  # 替换为你的API Key
)

# 定义工具列表
tools = [
    TavilySearchResults(
        max_results=5,
        description="搜索互联网获取最新信息。输入应该是搜索查询。"
    )
]

# 初始化Agent
# AgentType.CHAT_ZERO_SHOT_REACT_DECISION 是新版推荐的Agent类型
# 旧版是 AgentType.ZERO_SHOT_REACT_DESCRIPTION
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_ZERO_SHOT_REACT_DECISION,
    verbose=True,  # 开启详细日志，方便调试
    handle_parsing_errors=True  # LLM输出格式错误时自动处理
)

# 执行查询
result = agent.run("2024年诺贝尔物理学奖获得者是谁？他们的主要贡献是什么？")
print(result)

"""
Agent执行过程日志解读：
1. Thought: LLM分析用户问题，决定是否需要调用工具
2. Action: 调用哪个工具，以及输入参数
3. Observation: 工具执行返回的结果
4. （重复1-3直到认为可以回答）
5. Final Answer: 最终回答

完整日志输出类似：
> Entering new AgentExecutor chain...
Thought: 我需要搜索2024年诺贝尔物理学奖的相关信息。Action: tavily-search-input Action Input: 2024 Nobel Prize Physics winner
Observation: [结果包含获奖者姓名和贡献]
Thought: 我已经获取到了信息，现在可以回答用户问题了。Final Answer: 2024年诺贝尔物理学奖获得者是...
> Finished chain.
"""

2.3 构建多工具ReAct Agent

"""
生产级ReAct Agent实现
包含：天气查询、网页搜索、数学计算、日程查询、文件操作
"""

from typing import Annotated, Literal, Union
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain import hub
import json
import re
import requests
from datetime import datetime, timedelta

# ============================================
# 第一部分：定义工具函数（Tool Functions）
# ============================================

class Tools:
    """工具函数集合"""
    
    @staticmethod
    @tool
    def search_web(query: str) -> str:
        """
        搜索互联网获取最新信息。

        参数:
            query: 搜索查询语句，应包含足够的关键词以获得准确结果

        返回:
            搜索结果摘要列表
        """
        from duckduckgo_search import DDGS
        
        results = []
        with DDGS() as ddgs:
            for r in ddgs.news(query, max_results=5):
                results.append({
                    "title": r["title"],
                    "body": r["body"],
                    "url": r["url"]
                })
        
        if not results:
            return "未找到相关结果"
        
        # 格式化为字符串返回
        formatted = []
        for i, r in enumerate(results, 1):
            formatted.append(f"[{i}] {r['title']}\n  {r['body'][:200]}...\n  来源: {r['url']}")
        
        return "\n\n".join(formatted)
    
    @staticmethod
    @tool  
    def get_weather(city: str) -> str:
        """
        查询城市天气。

        参数:
            city: 城市名称，如"北京"、"上海"

        返回:
            天气信息描述
        """
        try:
            # 使用wttr.in服务（免费无需API Key）
            url = f"https://wttr.in/{city}?format=3"
            response = requests.get(url, timeout=10)
            
            if response.status_code == 200:
                return response.text.strip()
            else:
                return f"无法获取{city}的天气信息"
        except Exception as e:
            return f"查询天气失败: {str(e)}"
    
    @staticmethod
    @tool
    def calculate(expression: str) -> str:
        """
        执行安全数学计算。

        参数:
            expression: 数学表达式，如 "2 + 3 * 4" 或 "(10 + 5) / 2"

        返回:
            计算结果
        """
        try:
            # 使用numexpr进行安全计算（防止注入）
            import numexpr as ne
            
            # 安全检查：只允许数字和运算符
            if not re.match(r'^[\d\s\+\-\*\/\(\)\.\,]+$', expression):
                return f"表达式包含非法字符: {expression}"
            
            result = ne.evaluate(expression)
            return f"{expression} = {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"
    
    @staticmethod
    @tool
    def read_file(file_path: str, max_lines: int = 50) -> str:
        """
        读取文件内容。

        参数:
            file_path: 文件路径
            max_lines: 最大读取行数（防止读取超大文件）

        返回:
            文件内容前max_lines行
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                lines = []
                for i, line in enumerate(f):
                    if i >= max_lines:
                        return "\n".join(lines) + f"\n... (文件过大，已截断前{max_lines}行)"
                    lines.append(line.rstrip())
                return "\n".join(lines)
        except FileNotFoundError:
            return f"文件不存在: {file_path}"
        except Exception as e:
            return f"读取文件失败: {str(e)}"
    
    @staticmethod
    @tool
    def write_file(file_path: str, content: str) -> str:
        """
        写入内容到文件（覆盖）。

        参数:
            file_path: 文件路径
            content: 要写入的内容

        返回:
            操作结果
        """
        try:
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)
            return f"文件已成功写入: {file_path}"
        except Exception as e:
            return f"写入文件失败: {str(e)}"
    
    @staticmethod
    @tool
    def get_date(days_offset: int = 0) -> str:
        """
        获取当前日期或指定偏移后的日期。

        参数:
            days_offset: 相对今天偏移天数，正数为未来，负数为过去

        返回:
            格式化的日期字符串
        """
        target_date = datetime.now() + timedelta(days=days_offset)
        return target_date.strftime("%Y年%m月%d日 %A")
    
    @staticmethod
    @tool
    def get_system_info() -> str:
        """
        获取系统基本信息（操作系统、Python版本等）。
        无需输入参数。
        """
        import platform
        import sys
        
        info = {
            "操作系统": platform.system(),
            "系统版本": platform.release(),
            "Python版本": sys.version,
            "当前时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "CPU架构": platform.machine(),
            "主机名": platform.node()
        }
        
        return json.dumps(info, ensure_ascii=False, indent=2)

# ============================================
# 第二部分：构建Agent
# ============================================

class ReActAgent:
    """
    ReAct Agent主类
    
    工作流程：
    1. 用户输入 -> LLM分析 -> 判断是否需要工具
    2. 如果需要，选择工具并执行
    3. 工具结果返回给LLM -> 继续分析
    4. 直到LLM认为可以回答 -> 返回Final Answer
    """
    
    def __init__(self, api_key: str, model: str = "gpt-4o"):
        self.llm = ChatOpenAI(
            model=model,
            temperature=0,
            api_key=api_key
        )
        
        # 注册所有工具
        self.tools = [
            Tools.search_web,
            Tools.get_weather,
            Tools.calculate,
            Tools.read_file,
            Tools.write_file,
            Tools.get_date,
            Tools.get_system_info,
        ]
        
        # 从Hub获取提示词模板（LangChain提供了标准的ReAct提示词）
        # prompt = hub.pull("hwchase17/react-chat")
        # 如果想自己定义提示词，可以不用hub，直接用下面的
        prompt = self._build_custom_prompt()
        
        # 创建Agent
        self.agent = create_openai_functions_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )
        
        # 创建Agent执行器
        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            verbose=True,
            max_iterations=15,      # 最大迭代次数，防止无限循环
            max_execution_time=120, # 最大执行时间（秒）
            handle_parsing_errors=True
        )
    
    def _build_custom_prompt(self):
        """
        构建自定义提示词
        
        一个好的Agent提示词应该包含：
        1. Agent角色定义
        2. 可用工具说明
        3. 执行规则（何时用工具、如何回答）
        4. 输出格式要求
        """
        from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
        
        system_prompt = """你是一个智能助手，名字叫小云。你善于思考，能够通过调用工具来解决复杂问题。

## 可用工具

你可以通过以下工具来完成用户请求：

{tools}

## 工作规则

1. 如果用户询问的事实性问题，先尝试联网搜索
2. 如果需要进行计算，使用calculate工具
3. 如果需要读取文件，使用read_file工具
4. 如果需要写入文件，使用write_file工具
5. 如果需要日期，使用get_date工具
6. 如果需要天气，使用get_weather工具
7. 如果需要系统信息，使用get_system_info工具

## 执行流程

对于每个用户问题，按照以下步骤思考：

1. 分析用户问题，理解用户意图
2. 判断是否需要调用工具
3. 如果需要，调用最合适的工具
4. 观察工具返回结果
5. 如果结果足够，直接回答；如果不够，继续调用工具
6. 最终给出完整、准确的回答

## 输出要求

- 回答要简洁、有条理
- 如果调用了工具，明确说明你使用了什么工具
- 如果工具返回了数据，基于数据回答，而不是编造
- 不知道就说不知道，不要编造信息
"""
        
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=system_prompt),
            MessagesPlaceholder(variable_name="chat_history", optional=True),
            HumanMessage(content="{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")  # Agent的思考过程会记录在这里
        ])
        
        return prompt
    
    def run(self, query: str, chat_history: list = None) -> str:
        """
        执行Agent查询
        
        参数:
            query: 用户查询
            chat_history: 对话历史（可选，用于多轮对话）
        
        返回:
            Agent的回答
        """
        inputs = {"input": query}
        
        if chat_history:
            inputs["chat_history"] = chat_history
        
        result = self.agent_executor.invoke(inputs)
        
        return result["output"]
    
    def stream_run(self, query: str):
        """
        流式执行（实时看到思考过程）
        
        参数:
            query: 用户查询
        
        返回:
            生成器，逐步输出思考过程和最终答案
        """
        inputs = {"input": query}
        
        for step in self.agent_executor.stream(inputs):
            yield step

# ============================================
# 第三部分：使用示例
# ============================================

if __name__ == "__main__":
    # 初始化Agent
    agent = ReActAgent(
        api_key="sk-your-api-key-here",  # 替换为你的API Key
        model="gpt-4o"
    )
    
    # 单轮对话示例
    print("=" * 60)
    print("示例1：天气查询")
    print("=" * 60)
    result = agent.run("北京今天天气怎么样？需要穿什么衣服？")
    print(f"\n最终回答:\n{result}")
    
    print("\n" + "=" * 60)
    print("示例2：数学计算")
    print("=" * 60)
    result = agent.run("计算一下：如果一个投资每年收益率是8%，10年后10000元会变成多少？")
    print(f"\n最终回答:\n{result}")
    
    print("\n" + "=" * 60)
    print("示例3：复杂任务（多工具协作）")
    print("=" * 60)
    result = agent.run("帮我查一下上海今天的天气，然后把这个天气的摘要写入到/tmp/weather_report.txt文件中")
    print(f"\n最终回答:\n{result}")
    
    print("\n" + "=" * 60)
    print("示例4：系统信息查询")
    print("=" * 60)
    result = agent.run("查看一下当前系统的基本信息")
    print(f"\n最终回答:\n{result}")

2.4 构建Web服务（FastAPI部署）

"""
生产级Agent服务
使用FastAPI部署为RESTful API
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from datetime import datetime
import uvicorn
import os
import json
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="LangChain Agent API",
    description="生产级ReAct Agent服务",
    version="1.0.0"
)

# 全局Agent实例（启动时初始化）
agent_instance = None

class ChatRequest(BaseModel):
    """聊天请求模型"""
    message: str = Field(..., description="用户消息")
    session_id: Optional[str] = Field(None, description="会话ID，用于多轮对话")
    model: str = Field(default="gpt-4o", description="使用的模型")

class ChatResponse(BaseModel):
    """聊天响应模型"""
    answer: str = Field(..., description="Agent回答")
    session_id: str = Field(..., description="会话ID")
    tools_used: List[str] = Field(default=[], description="调用的工具列表")
    tokens_used: Optional[int] = Field(None, description="token使用量")
    execution_time_ms: int = Field(..., description="执行耗时（毫秒）")

class HealthResponse(BaseModel):
    """健康检查响应"""
    status: str
    timestamp: str
    version: str

# 会话存储（生产环境应该用Redis）
sessions: Dict[str, List[Dict]] = {}

def get_agent():
    """获取或创建Agent实例"""
    global agent_instance
    
    if agent_instance is None:
        from your_agent_module import ReActAgent  # 导入你自己的Agent类
        
        api_key = os.environ.get("OPENAI_API_KEY")
        if not api_key:
            raise RuntimeError("OPENAI_API_KEY环境变量未设置")
        
        agent_instance = ReActAgent(api_key=api_key)
    
    return agent_instance

@app.post("/chat", response_model=ChatResponse)
def chat(request: ChatRequest):
    """
    主聊天接口
    """
    import time
    start_time = time.time()
    
    try:
        agent = get_agent()
        
        # 获取或创建会话
        session_id = request.session_id or "default"
        chat_history = sessions.get(session_id, [])
        
        # 执行查询
        logger.info(f"[{session_id}] 用户: {request.message}")
        
        answer = agent.run(
            query=request.message,
            chat_history=chat_history
        )
        
        # 更新会话历史
        chat_history.append({"role": "user", "content": request.message})
        chat_history.append({"role": "assistant", "content": answer})
        sessions[session_id] = chat_history
        
        execution_time = int((time.time() - start_time) * 1000)
        
        logger.info(f"[{session_id}] Agent: {answer[:100]}... ({execution_time}ms)")
        
        return ChatResponse(
            answer=answer,
            session_id=session_id,
            tools_used=[],  # 从agent返回结果中提取
            execution_time_ms=execution_time
        )
    
    except Exception as e:
        logger.error(f"聊天请求失败: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.delete("/chat/{session_id}")
def delete_session(session_id: str):
    """
    删除会话（清理历史）
    """
    if session_id in sessions:
        del sessions[session_id]
        return {"status": "deleted", "session_id": session_id}
    else:
        raise HTTPException(status_code=404, detail="会话不存在")

@app.get("/health", response_model=HealthResponse)
def health():
    """
    健康检查接口
    """
    return HealthResponse(
        status="healthy",
        timestamp=datetime.now().isoformat(),
        version="1.0.0"
    )

@app.get("/sessions")
def list_sessions():
    """
    列出所有活跃会话（调试用）
    """
    return {
        "count": len(sessions),
        "sessions": list(sessions.keys())
    }

# 启动命令
if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        workers=4,  # 生产环境用多进程
        log_level="info"
    )

2.5 使用PM2管理服务

# 创建PM2配置文件
cat > ecosystem.config.js << 'EOF'
module.exports = {
  apps: [{
    name: 'langchain-agent-api',
    script: './agent_server.py',
    cwd: '/opt/agent',
    env: {
      NODE_ENV: 'production',
      OPENAI_API_KEY: 'sk-your-api-key-here',  // 生产环境建议用环境变量
    },
    instances: 4,                    // 4个进程，充分利用CPU多核
    exec_mode: 'cluster',            // 集群模式
    autorestart: true,
    watch: false,
    max_memory_restart: '1G',
    log_date_format: 'YYYY-MM-DD HH:mm:ss',
    error_file: './logs/error.log',
    out_file: './logs/out.log',
    time: true,
    // 日志轮转
    pm2_log_date_format: 'YYYY-MM-DD',
    rotateInterval: '1 0 * * *',     // 每天0点轮转
    rotateMaxSize: '10M',
    rotateMaxFiles: 10,
  }]
};
EOF

# 创建日志目录
mkdir -p logs

# 启动服务
pm2 start ecosystem.config.js

# 查看状态
pm2 list
pm2 logs langchain-agent-api --lines 50

# 常用命令
pm2 restart langchain-agent-api   # 重启
pm2 stop langchain-agent-api      # 停止
pm2 delete langchain-agent-api    # 删除

# 开机自启
pm2 save
pm2 startup

2.6 常见错误排查

三、效果：Agent系统性能评估

3.1 功能覆盖度

3.2 端到端性能

四、总结：Agent开发核心经验

4.1 提示词工程要点

1. 工具描述要清晰：每个工具的description要明确说明用途和参数格式
2. 执行规则要明确：告诉Agent何时用什么工具，如何处理结果
3. 输出格式要约束：要求明确"不知道就说不知道"而不是编造
4. 示例有帮助：few-shot示例可以显著提升工具调用准确率

4.2 生产环境注意事项

• 安全：敏感操作加确认：文件写入、删除等操作需要二次确认
• 限流：防止滥用：对单用户单IP做限流
• 监控：接入LangSmith：方便追踪Agent的执行过程和问题排查
• 降级：优雅处理失败：工具调用失败时要有降级策略
• 成本控制：设置max_tokens防止LLM生成过长回答

# Docker一键部署
cat > Dockerfile << 'EOF'
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 运行
CMD ["python", "agent_server.py"]
EOF

# requirements.txt内容
# langchain>=0.2.0
# langchain-openai>=0.1.0
# langchain-community>=0.2.0
# fastapi>=0.110.0
# uvicorn>=0.27.0
# duckduckgo-search>=4.0.0
# requests>=2.31.0
# numexpr>=2.8.0
# python-dateutil>=2.8.0

# 构建和运行
docker build -t langchain-agent .
docker run -d -p 8000:8000 \
    -e OPENAI_API_KEY=$OPENAI_API_KEY \
    --name langchain-agent \
    langchain-agent

4.3 快速命令汇总

# 环境搭建
conda create -n langchain python=3.11 -y && conda activate langchain
pip install langchain langchain-openai openai langchain-community

# 测试Agent
python -c "from langchain.agents import initialize_agent; print('OK')"

# 启动API服务
python agent_server.py

# 用PM2管理
pm2 start ecosystem.config.js
pm2 list && pm2 logs

# Docker部署
docker build -t langchain-agent .
docker run -d -p 8000:8000 -e OPENAI_API_KEY=$OPENAI_API_KEY langchain-agent

# API测试
curl -X POST http://localhost:8000/chat \
    -H "Content-Type: application/json" \
    -d '{"message": "北京天气怎么样？"}'