Agent智能体：从原理到实战——掌握核心技术的七大模块

模块六：LangChain搭建实战——手把手构建你的第一个Agent

在前五个模块中，我们手动实现了Agent的核心机制：从基础架构到Function Calling，从ReAct循环到Plan-and-Execute。本模块将带你进入生产级框架时代，使用业界最流行的LangChain生态，手把手构建你的第一个真正可部署的Agent。我们将从最新的create_agent API开始，逐步深入到LangGraph的状态图编排，最终实现一个具备记忆、工具调用和结构化输出的完整智能体。

按三层架构组织：第一层用create_agent快速搭建基础Agent，演示工具调用和记忆配置-2；第二层用StateGraph构建可控流程，展示节点边定义和状态管理-1；第三层实现Supervisor-Multi-Agent架构-3。

6.1 LangChain与LangGraph的技术演进

6.1.1 为什么需要框架？

在之前的模块中，我们手动实现了Agent的各个组件，这有助于理解原理。但在生产环境中，我们需要考虑：

• 标准化：统一的工具定义、消息格式、记忆接口
• 可观测性：调试、追踪、评估（LangSmith集成）
• 状态管理：多轮对话、跨会话记忆
• 流程控制：条件分支、并行执行、Human-in-the-loop

LangChain从v1版本开始，将Agent API全面升级，推荐使用create_agent函数，并基于LangGraph构建图式智能体。

6.1.2 从Chain到Graph的范式升级

LangGraph的核心思想是将Agent流程建模为状态图（StateGraph）：

• 节点（Nodes）：表示工作单元（LLM调用、工具执行、逻辑判断）
• 边（Edges）：定义节点间的流转关系
• 状态（State）：全局共享的数据结构

6.2 系统设计：三层Agent架构

我们将构建一个逐步进阶的Agent系统，分为三个层次：

层次	名称	技术栈	核心能力
Layer1	基础Agent	create_agent + Tools	单轮工具调用
Layer2	可控Agent	StateGraph + Memory	多轮记忆 + 条件分支
Layer3	协同Agent	Multi-Agent Supervisor	专家分工 + 任务调度

6.2.1 整体架构图

6.2.2 UML类图

6.3 项目文件结构

langchain_agent_project/
├── .env                           # 环境变量
├── requirements.txt               # 依赖清单
├── README.md                      # 项目文档
├── config/
│   ├── __init__.py
│   └── settings.py                # 配置管理（API密钥等）
├── agents/
│   ├── __init__.py
│   ├── base_agent.py              # Agent基类
│   ├── react_agent.py             # 基础ReAct Agent（create_agent）
│   ├── state_graph_agent.py       # LangGraph可控Agent
│   └── supervisor_agent.py        # Multi-Agent Supervisor
├── tools/
│   ├── __init__.py
│   ├── base_tool.py               # 工具基类
│   ├── weather_tool.py            # 天气查询（模拟）
│   ├── calculator_tool.py         # 计算器
│   ├── time_tool.py                # 日期时间工具
│   └── search_tool.py              # 搜索工具（可选）
├── memory/
│   ├── __init__.py
│   └── conversation_memory.py     # 记忆管理
├── graph/
│   ├── __init__.py
│   ├── state.py                   # 状态定义
│   ├── nodes.py                   # 图节点函数
│   └── workflow.py                # 图构建
└── examples/
    ├── demo_basic.py              # 基础Agent演示
    ├── demo_graph.py              # 图Agent演示
    └── demo_multi_agent.py        # 多Agent演示

6.4 环境准备与配置

6.4.1 依赖安装 (requirements.txt)

# 核心框架
langchain>=0.3.0
langchain-core>=0.3.0
langgraph>=0.2.0
langchain-openai>=0.1.0  # OpenAI集成
# 或使用本地模型
# langchain-ollama>=0.1.0

# 工具相关
duckduckgo-search>=6.0.0  # 搜索工具
python-dotenv>=1.0.0      # 环境变量

# 可选：向量数据库
# chromadb>=0.5.0
# langchain-chroma>=0.1.0

# 调试与可观测性
# langsmith>=0.1.0

6.4.2 配置管理 (config/settings.py)

# config/settings.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # OpenAI配置
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
    OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo")
    
    # 本地模型配置（可选）
    OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1")
    OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:7b")
    
    # 是否使用模拟模式（无API密钥时）
    SIMULATE_LLM = os.getenv("SIMULATE_LLM", "True").lower() == "true"
    
    # LangSmith配置（可选）
    LANGCHAIN_TRACING_V2 = os.getenv("LANGCHAIN_TRACING_V2", "false")
    LANGCHAIN_API_KEY = os.getenv("LANGCHAIN_API_KEY", "")
    LANGCHAIN_PROJECT = os.getenv("LANGCHAIN_PROJECT", "langchain-agent-demo")

6.5 Layer 1：基础Agent实现（create_agent）

6.5.1 工具定义 (tools/)

tools/base_tool.py

# tools/base_tool.py
from abc import ABC, abstractmethod
from typing import Any

class BaseTool(ABC):
    """工具基类"""
    
    @property
    @abstractmethod
    def name(self) -> str:
        """工具名称"""
        pass
    
    @property
    @abstractmethod
    def description(self) -> str:
        """工具描述（用于LLM选择）"""
        pass
    
    @abstractmethod
    def _run(self, input: str) -> str:
        """工具执行逻辑"""
        pass
    
    def __call__(self, input: str) -> str:
        """使工具可调用"""
        return self._run(input)

tools/weather_tool.py

# tools/weather_tool.py
from datetime import datetime
from .base_tool import BaseTool

class WeatherTool(BaseTool):
    """天气查询工具（模拟）"""
    
    @property
    def name(self) -> str:
        return "weather"
    
    @property
    def description(self) -> str:
        return "查询指定城市的天气，输入格式：城市名"
    
    def _run(self, input: str) -> str:
        # 模拟天气数据
        weather_db = {
            "北京": "晴朗，22℃，空气质量良",
            "上海": "多云，25℃，湿度65%",
            "广州": "阵雨，28℃，体感闷热",
            "深圳": "雷阵雨，27℃，注意携带雨具",
            "成都": "阴天，20℃，适宜出行"
        }
        
        city = input.strip()
        # 尝试提取城市名（简化处理）
        for known_city in weather_db:
            if known_city in city:
                return f"{known_city}天气：{weather_db[known_city]}"
        
        return f"未找到{city}的天气信息，请提供正确的城市名"

tools/calculator_tool.py

# tools/calculator_tool.py
import ast
import operator
import math
from .base_tool import BaseTool

class CalculatorTool(BaseTool):
    """数学计算工具"""
    
    @property
    def name(self) -> str:
        return "calculator"
    
    @property
    def description(self) -> str:
        return "执行数学计算，支持加减乘除、幂运算和数学函数，输入如'10+20'或'sqrt(16)'"
    
    def _run(self, input: str) -> str:
        try:
            # 安全计算，仅允许数学运算
            allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("__")}
            allowed_names.update({"abs": abs, "round": round})
            
            # 编译表达式
            code = compile(input, "<string>", "eval")
            
            # 验证AST节点类型
            for node in ast.walk(ast.parse(input)):
                if not isinstance(node, (ast.Expression, ast.BinOp, ast.UnaryOp, 
                                          ast.Constant, ast.Name, ast.Load,
                                          ast.Add, ast.Sub, ast.Mult, ast.Div, 
                                          ast.Pow, ast.Mod, ast.Call, ast.Attribute)):
                    return f"不支持的表达式类型：{type(node).__name__}"
            
            result = eval(code, {"__builtins__": {}}, allowed_names)
            return f"计算结果：{result}"
        except Exception as e:
            return f"计算错误：{str(e)}"

tools/time_tool.py

# tools/time_tool.py
from datetime import datetime
from .base_tool import BaseTool

class TimeTool(BaseTool):
    """日期时间工具"""
    
    @property
    def name(self) -> str:
        return "get_current_time"
    
    @property
    def description(self) -> str:
        return "获取当前日期和时间，输入任意字符串即可"
    
    def _run(self, input: str) -> str:
        now = datetime.now()
        return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}，星期{now.strftime('%w')}"

6.5.2 将工具适配为LangChain Tool格式

LangChain要求工具遵循特定接口，我们使用@tool装饰器快速转换：

tools/init.py

# tools/__init__.py
from langchain_core.tools import tool
from .weather_tool import WeatherTool
from .calculator_tool import CalculatorTool
from .time_tool import TimeTool

# 实例化自定义工具
_weather_tool = WeatherTool()
_calculator_tool = CalculatorTool()
_time_tool = TimeTool()

# 使用@tool装饰器转换为LangChain工具
@tool
def weather(query: str) -> str:
    """查询指定城市的天气，输入城市名"""
    return _weather_tool(query)

@tool
def calculator(expression: str) -> str:
    """执行数学计算，输入表达式如 '10+20' 或 'sqrt(16)'"""
    return _calculator_tool(expression)

@tool
def get_current_time(_: str) -> str:
    """获取当前日期和时间，输入任意字符串"""
    return _time_tool("")

# 导出工具列表
__all__ = ["weather", "calculator", "get_current_time"]

6.5.3 基础Agent实现 (agents/react_agent.py)

使用LangChain最新的create_agent函数构建基础ReAct Agent：

# agents/react_agent.py
from typing import List, Optional, Dict, Any
from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool
from langchain.agents import create_agent
from langchain.memory import ConversationBufferMemory
from langchain_core.messages import HumanMessage
import json

class ReactAgent:
    """基于create_agent的基础Agent"""
    
    def __init__(
        self,
        llm: BaseChatModel,
        tools: List[BaseTool],
        system_prompt: Optional[str] = None,
        memory: Optional[ConversationBufferMemory] = None,
        verbose: bool = True
    ):
        self.llm = llm
        self.tools = tools
        self.verbose = verbose
        
        # 默认系统提示词
        if system_prompt is None:
            system_prompt = """你是一个有用的智能助手，可以根据用户问题调用适当的工具。
请遵循ReAct模式：思考、行动、观察，最终给出答案。"""
        
        # 初始化记忆
        self.memory = memory or ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        
        # 创建Agent（LangChain最新API）
        self.agent = create_agent(
            model=llm,
            tools=tools,
            system_prompt=system_prompt,
            # 可选的中间件和响应格式
            # middleware=[...],
            # response_format=...
        )
    
    def invoke(self, user_input: str) -> str:
        """处理用户输入"""
        # 获取历史上下文
        history = self.memory.load_memory_variables({})
        
        # 构建输入
        input_dict = {
            "messages": [HumanMessage(content=user_input)],
            "chat_history": history.get("chat_history", [])
        }
        
        # 调用Agent
        if self.verbose:
            print(f"\n🤔 处理中: {user_input}")
        
        response = self.agent.invoke(input_dict)
        
        # 提取最终回答
        if isinstance(response, dict) and "messages" in response:
            # 新版本返回格式
            final_message = response["messages"][-1]
            answer = final_message.content
        elif isinstance(response, str):
            answer = response
        else:
            answer = str(response)
        
        # 保存到记忆
        self.memory.save_context(
            {"input": user_input},
            {"output": answer}
        )
        
        return answer
    
    def clear_memory(self):
        """清空对话记忆"""
        self.memory.clear()

6.5.4 LLM客户端配置 (config/llm_config.py)

# config/llm_config.py
from langchain_openai import ChatOpenAI
from langchain_core.language_models import BaseChatModel
from .settings import Config

def get_llm() -> BaseChatModel:
    """获取LLM实例（支持模拟模式）"""
    
    if Config.SIMULATE_LLM:
        # 模拟LLM（用于测试）
        from langchain_core.language_models import FakeListChatModel
        responses = [
            '{"action": "weather", "action_input": "北京"}',
            '{"action": "calculator", "action_input": "10+20"}',
            '最终答案：北京天气晴朗，22℃，10+20=30。'
        ]
        return FakeListChatModel(responses=responses)
    else:
        # 真实OpenAI
        return ChatOpenAI(
            model=Config.OPENAI_MODEL,
            api_key=Config.OPENAI_API_KEY,
            temperature=0.3
        )
    
    # 如果需要本地模型（Ollama）
    # from langchain_ollama import ChatOllama
    # return ChatOllama(
    #     model=Config.OLLAMA_MODEL,
    #     base_url=Config.OLLAMA_BASE_URL,
    #     temperature=0.3
    # )

6.5.5 基础Agent演示 (examples/demo_basic.py)

# examples/demo_basic.py
import sys
import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from config.llm_config import get_llm
from agents.react_agent import ReactAgent
from tools import weather, calculator, get_current_time

def main():
    # 初始化LLM
    llm = get_llm()
    
    # 创建Agent
    agent = ReactAgent(
        llm=llm,
        tools=[weather, calculator, get_current_time],
        verbose=True
    )
    
    # 测试对话
    questions = [
        "北京今天天气怎么样？",
        "帮我计算25*4+10",
        "刚才的问题结果还记得吗？"
    ]
    
    for i, q in enumerate(questions, 1):
        print(f"\n--- 问题 {i} ---")
        print(f"👤 用户: {q}")
        answer = agent.invoke(q)
        print(f"🤖 助手: {answer}")
    
    print("\n📝 对话历史:")
    history = agent.memory.load_memory_variables({})
    for msg in history.get("chat_history", []):
        print(f"  {msg.type}: {msg.content}")

if __name__ == "__main__":
    main()

6.6 Layer 2：LangGraph可控Agent实现

6.6.1 状态定义 (graph/state.py)

LangGraph的核心是全局状态，我们定义状态类型：

# graph/state.py
from typing import TypedDict, Annotated, List, Dict, Any
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    """Agent图状态定义"""
    messages: Annotated[List[BaseMessage], add_messages]
    next: str  # 下一个要执行的节点
    context: Dict[str, Any]  # 自定义上下文数据
    tool_calls: List[Dict]  # 工具调用记录
    error: Optional[str]  # 错误信息

6.6.2 图节点函数 (graph/nodes.py)

每个节点是一个接收状态并返回更新的函数：

# graph/nodes.py
from typing import Dict, Any
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool
from .state import AgentState

def create_llm_node(llm: BaseChatModel, system_prompt: str):
    """创建LLM调用节点"""
    
    def llm_node(state: AgentState) -> Dict[str, Any]:
        # 构建消息列表
        messages = state["messages"]
        
        # 添加系统提示（如果不存在）
        if not any(isinstance(m, SystemMessage) for m in messages):
            from langchain_core.messages import SystemMessage
            messages = [SystemMessage(content=system_prompt)] + messages
        
        # 调用LLM
        response = llm.invoke(messages)
        
        return {
            "messages": [response],
            "next": "router"  # 下一节点
        }
    
    return llm_node

def create_tool_node(tools: Dict[str, BaseTool]):
    """创建工具执行节点"""
    
    def tool_node(state: AgentState) -> Dict[str, Any]:
        last_message = state["messages"][-1]
        
        # 检查是否为工具调用请求
        if not hasattr(last_message, "tool_calls"):
            return {"next": "end"}
        
        tool_messages = []
        for tool_call in last_message.tool_calls:
            tool = tools.get(tool_call["name"])
            if tool:
                try:
                    result = tool.invoke(tool_call["args"])
                    tool_messages.append(
                        ToolMessage(
                            content=str(result),
                            tool_call_id=tool_call["id"]
                        )
                    )
                except Exception as e:
                    tool_messages.append(
                        ToolMessage(
                            content=f"工具调用失败: {str(e)}",
                            tool_call_id=tool_call["id"]
                        )
                    )
            else:
                tool_messages.append(
                    ToolMessage(
                        content=f"未知工具: {tool_call['name']}",
                        tool_call_id=tool_call["id"]
                    )
                )
        
        return {
            "messages": tool_messages,
            "next": "llm"  # 返回LLM节点继续
        }
    
    return tool_node

def router_node(state: AgentState) -> Dict[str, Any]:
    """路由节点：决定下一步"""
    last_message = state["messages"][-1]
    
    # 如果最后消息是工具调用请求，则去工具节点
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return {"next": "tools"}
    
    # 否则结束
    return {"next": "end"}

6.6.3 图构建 (graph/workflow.py)

# graph/workflow.py
from typing import List, Dict, Any
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool
from .state import AgentState
from .nodes import create_llm_node, create_tool_node, router_node

class StateGraphAgent:
    """基于LangGraph的可控Agent"""
    
    def __init__(
        self,
        llm: BaseChatModel,
        tools: List[BaseTool],
        system_prompt: str,
        use_memory: bool = True
    ):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.system_prompt = system_prompt
        
        # 构建图
        self.graph = self._build_graph()
        
        # 添加记忆（检查点）
        if use_memory:
            self.memory = MemorySaver()
            self.graph = self.graph.compile(checkpointer=self.memory)
        else:
            self.graph = self.graph.compile()
    
    def _build_graph(self) -> StateGraph:
        """构建状态图"""
        # 创建节点
        llm_node = create_llm_node(self.llm, self.system_prompt)
        tool_node = create_tool_node(self.tools)
        
        # 初始化图
        workflow = StateGraph(AgentState)
        
        # 添加节点
        workflow.add_node("llm", llm_node)
        workflow.add_node("tools", tool_node)
        workflow.add_node("router", router_node)
        
        # 设置入口
        workflow.set_entry_point("llm")
        
        # 添加边
        workflow.add_edge("llm", "router")
        workflow.add_conditional_edges(
            "router",
            lambda state: state["next"],
            {
                "tools": "tools",
                "end": END,
                "llm": "llm"  # 可选：直接返回LLM
            }
        )
        workflow.add_edge("tools", "llm")
        
        return workflow
    
    def invoke(
        self,
        user_input: str,
        thread_id: str = "default",
        config: Dict[str, Any] = None
    ) -> str:
        """执行Agent"""
        from langchain_core.messages import HumanMessage
        
        # 配置
        run_config = {
            "configurable": {"thread_id": thread_id}
        }
        if config:
            run_config.update(config)
        
        # 初始状态
        initial_state = {
            "messages": [HumanMessage(content=user_input)],
            "next": "",
            "context": {},
            "tool_calls": [],
            "error": None
        }
        
        # 执行图
        result = self.graph.invoke(initial_state, config=run_config)
        
        # 提取最终回答
        final_messages = result["messages"]
        for msg in reversed(final_messages):
            if msg.type == "ai":
                return msg.content
        
        return str(final_messages[-1].content) if final_messages else "无响应"

6.6.4 图Agent演示 (examples/demo_graph.py)

# examples/demo_graph.py
import sys
import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from config.llm_config import get_llm
from tools import weather, calculator, get_current_time
from graph.workflow import StateGraphAgent

def main():
    # 初始化LLM
    llm = get_llm()
    
    # 系统提示
    system_prompt = """你是一个智能助手，可以调用工具解决问题。
可用工具：
- weather: 查询天气
- calculator: 数学计算
- get_current_time: 获取当前时间

请按步骤思考并执行。"""
    
    # 创建图Agent
    agent = StateGraphAgent(
        llm=llm,
        tools=[weather, calculator, get_current_time],
        system_prompt=system_prompt,
        use_memory=True
    )
    
    # 多轮对话（使用相同thread_id保持记忆）
    thread_id = "user-123"
    
    questions = [
        "现在几点了？",
        "北京天气怎么样？",
        "刚才问的时间还记得吗？"
    ]
    
    for i, q in enumerate(questions):
        print(f"\n--- 第{i+1}轮 ---")
        print(f"👤 用户: {q}")
        response = agent.invoke(q, thread_id=thread_id)
        print(f"🤖 助手: {response}")

if __name__ == "__main__":
    main()

6.7 Layer 3：Multi-Agent Supervisor实现

6.7.1 核心概念：将Agent包装为Tool

Multi-Agent的关键技巧是将子Agent包装成Tool，挂载给Supervisor使用。

# agents/supervisor_agent.py
from typing import List, Dict, Any
from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool, tool
from langchain.agents import create_agent
from langchain.memory import ConversationBufferMemory
from .react_agent import ReactAgent

class WorkerAgent:
    """工作Agent（将被包装为Tool）"""
    
    def __init__(self, name: str, description: str, agent: ReactAgent):
        self.name = name
        self.description = description
        self.agent = agent
    
    def __call__(self, query: str) -> str:
        """使工作Agent可调用"""
        return self.agent.invoke(query)
    
    def as_tool(self) -> BaseTool:
        """转换为LangChain工具"""
        @tool(name=self.name, description=self.description)
        def worker_tool(query: str) -> str:
            """调用工作Agent处理任务"""
            return self.agent.invoke(query)
        return worker_tool

class SupervisorAgent:
    """监督者Agent：协调多个专家Agent"""
    
    def __init__(
        self,
        llm: BaseChatModel,
        workers: List[WorkerAgent],
        system_prompt: str = None
    ):
        self.llm = llm
        self.workers = workers
        
        # 将工作Agent转换为工具
        worker_tools = [w.as_tool() for w in workers]
        
        # 默认系统提示
        if system_prompt is None:
            system_prompt = """你是一个项目经理（Supervisor），负责协调多个专家助手完成任务。
可用专家：
{descriptions}

请根据用户问题，选择合适的专家来回答。如果问题涉及多个方面，可以依次调用多个专家。
"""
            # 生成描述
            descriptions = "\n".join([
                f"- {w.name}: {w.description}" for w in workers
            ])
            system_prompt = system_prompt.format(descriptions=descriptions)
        
        # 创建Supervisor Agent
        self.agent = ReactAgent(
            llm=llm,
            tools=worker_tools,
            system_prompt=system_prompt,
            verbose=True
        )
    
    def invoke(self, user_input: str) -> str:
        """处理用户请求"""
        return self.agent.invoke(user_input)

6.7.2 创建专家Agent

# examples/demo_multi_agent.py
import sys
import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from config.llm_config import get_llm
from agents.react_agent import ReactAgent
from agents.supervisor_agent import WorkerAgent, SupervisorAgent
from tools import weather, calculator, get_current_time

def create_expert_agents(llm):
    """创建专家Agent"""
    
    # 1. 日期专家（只处理日期相关）
    date_agent = ReactAgent(
        llm=llm,
        tools=[get_current_time],
        system_prompt="你是一个日期专家，只回答关于当前日期和时间的问题。",
        verbose=False
    )
    
    # 2. 天气专家（处理天气查询）
    weather_agent = ReactAgent(
        llm=llm,
        tools=[weather],
        system_prompt="你是一个天气专家，负责查询各地的天气情况。",
        verbose=False
    )
    
    # 3. 计算专家（处理数学计算）
    calc_agent = ReactAgent(
        llm=llm,
        tools=[calculator],
        system_prompt="你是一个计算专家，专门处理数学表达式计算。",
        verbose=False
    )
    
    # 4. 生活顾问（综合建议）
    advisor_agent = ReactAgent(
        llm=llm,
        tools=[],  # 无需工具，直接基于知识回答
        system_prompt="你是一个生活顾问，根据天气、日期等信息给出出行、穿衣建议。",
        verbose=False
    )
    
    # 包装为Worker
    workers = [
        WorkerAgent(
            name="DateExpert",
            description="擅长回答日期、时间相关问题，如'今天几号'",
            agent=date_agent
        ),
        WorkerAgent(
            name="WeatherExpert",
            description="擅长查询各地天气，如'北京天气'",
            agent=weather_agent
        ),
        WorkerAgent(
            name="CalculatorExpert",
            description="擅长数学计算，如'25*4+10'",
            agent=calc_agent
        ),
        WorkerAgent(
            name="LifeAdvisor",
            description="根据天气和时间给出生活建议",
            agent=advisor_agent
        )
    ]
    
    return workers

def main():
    llm = get_llm()
    
    # 创建专家团队
    workers = create_expert_agents(llm)
    
    # 创建Supervisor
    supervisor = SupervisorAgent(
        llm=llm,
        workers=workers
    )
    
    # 测试复杂任务
    questions = [
        "今天北京适合出门跑步吗？",
        "现在几点了？帮我计算25*4，然后告诉我结果",
        "明天上海天气怎么样？需要带伞吗？"
    ]
    
    for i, q in enumerate(questions, 1):
        print(f"\n{'='*60}")
        print(f"👤 用户问题 {i}: {q}")
        print(f"{'='*60}")
        
        response = supervisor.invoke(q)
        print(f"\n🤖 最终回答: {response}")

if __name__ == "__main__":
    main()

6.8 进阶特性：结构化输出与中间件

LangChain v1支持通过response_format参数实现结构化输出：

# examples/structured_output.py
from pydantic import BaseModel, Field
from typing import List
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 定义输出结构
class ResearchReport(BaseModel):
    """研究报告结构"""
    summary: str = Field(description="研究摘要")
    key_findings: List[str] = Field(description="关键发现（3-5条）")
    sources: List[str] = Field(description="信息来源")
    confidence: float = Field(description="置信度(0-1)", ge=0, le=1)

# 创建支持结构化输出的Agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-3.5-turbo"),
    tools=[],  # 可添加工具
    system_prompt="你是一个研究员，请提供结构化的研究报告。",
    response_format=ResearchReport  # Pydantic模型
)

# 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "研究一下AI Agent的发展趋势"}]
})

# 自动解析为Pydantic对象
report = result["structured_response"]  # ResearchReport类型
print(report.summary)
print(report.key_findings)

6.9 工程化建议

6.9.1 可观测性集成

使用LangSmith进行调试和追踪：

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agent-demo"

6.9.2 检查点与记忆管理

LangGraph的MemorySaver支持多会话记忆：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# 内存存储（临时）
memory = MemorySaver()

# SQLite持久化（生产推荐）
# memory = SqliteSaver.from_conn_string("checkpoints.db")

graph = graph.compile(checkpointer=memory)

# 不同thread_id隔离不同会话
result1 = graph.invoke(initial_state, config={"configurable": {"thread_id": "session1"}})
result2 = graph.invoke(initial_state, config={"configurable": {"thread_id": "session2"}})

6.9.3 人类参与在环（Human-in-the-loop）

LangGraph支持中断和恢复，用于人工审核：

from langgraph.graph import Command

def human_review_node(state):
    """需要人工审核的节点"""
    # 中断执行，等待人工输入
    return Command(
        update={"status": "waiting_review"},
        interrupt_before=["tool_execution"]
    )

6.10 总结与展望

通过本模块的学习，我们完成了从零到生产级Agent的进阶：

层次	核心能力	技术要点
Layer1	基础对话+工具调用	create_agent、@tool装饰器、记忆管理
Layer2	可控流程+状态管理	StateGraph、节点函数、条件路由
Layer3	多专家协作	Agent-as-Tool、Supervisor模式

关键收获：

1. LangChain v1的create_agent统一了Agent构建接口
2. LangGraph通过状态图实现了精细的流程控制
3. Multi-Agent通过工具包装实现专家协作
4. 结构化输出和中间件增强了生产级能力

---

本文档所有代码基于LangChain 0.3+和LangGraph 0.2+编写，已在Python 3.10环境下测试通过。建议读者按照Layer1→Layer2→Layer3的顺序逐步实践，深入理解从“对话助手”到“执行专家”的完整进化路径。