第 7 章：Agent 框架实战

本章深入 Agent 框架的核心概念与实战应用。如果说上一章是学会"调用 API"，本章就是学会"编排智能"——让 LLM 像前端组件一样被组合、复用和调度，构建真正能解决复杂问题的 AI 系统。

7.1 Agent 概念深化

Agent（智能体）是 AI 领域最热门的概念之一。但到底什么是 Agent？它和直接调用 LLM API 有什么区别？

Agent vs 普通 LLM 调用

普通 LLM 调用是"一问一答"：你提问题，模型给答案。而 Agent 是"自主执行"：你给目标，Agent 自己规划步骤、调用工具、观察结果、调整策略，直到完成任务。

维度	普通 LLM 调用	Agent
交互模式	单轮/多轮对话	规划-执行-观察循环
工具使用	需手动处理	自动选择并调用
错误处理	人工介入	自动重试、调整
记忆	固定上下文	长期记忆、向量检索
适用场景	问答、生成	复杂任务、自动化流程

前端类比

普通 LLM 调用像静态页面——用户请求什么就返回什么；Agent 像完整的 Web 应用——有状态管理（Memory）、路由系统（Planning）、API 调用（Tools）和错误边界（Error Handling）。

规划-执行-观察循环

Agent 的核心工作模式是一个循环：

1. 规划（Planning） Agent 分析目标，制定执行计划。例如"查天气"任务可能被分解为：确定城市 → 调用天气 API → 格式化结果。
2. 执行（Action） 按计划调用工具或执行代码。Agent 可能调用搜索引擎、数据库、计算器等外部能力。
3. 观察（Observation） 收集工具返回的结果，评估是否达到目标。如果未完成，进入下一轮规划。

ReAct 模式

ReAct（Reasoning + Acting）是当前最流行的 Agent 设计模式。它要求模型在每一步同时输出"思考过程"和"执行动作"，形成交错推理。

用户：2024 年诺贝尔物理学奖得主是谁？他出生在哪一年？

思考 1：我需要查找 2024 年诺贝尔物理学奖得主的信息。
动作 1：search("2024 Nobel Prize Physics winner")
观察 1：2024 年诺贝尔物理学奖授予 John J. Hopfield 和 Geoffrey E. Hinton。

思考 2：我需要分别查找两位科学家的出生年份。
动作 2：search("John J. Hopfield born")
观察 2：John J. Hopfield 出生于 1933 年 7 月 15 日。

思考 3：现在查找 Geoffrey Hinton 的出生年份。
动作 3：search("Geoffrey Hinton born")
观察 3：Geoffrey Everest Hinton 出生于 1947 年 12 月 6 日。

思考 4：我已经获得了所有需要的信息，可以回答用户了。
动作 4：finish("2024 年诺贝尔物理学奖得主是 John J. Hopfield（1933 年出生）和 Geoffrey E. Hinton（1947 年出生）。")

7.2 LangChain 核心

LangChain 是 Python 生态中最流行的 LLM 应用框架。它的设计理念与前端工程化非常相似：组件化、可组合、可扩展。

前端类比

LangChain 就像 LLM 界的 React：提供组件（Components）、状态管理（Memory）、路由（Chains/Routers）和工具集成（Tools），让你像搭积木一样构建 AI 应用。

安装 LangChain

LangChain 0.3+ 采用模块化架构，根据需求安装对应的子包：

# 核心依赖
pip install langchain-core langchain-openai langchain-community

# 如需使用 LangGraph（状态机工作流，7.8 节介绍）
pip install langgraph

版本说明

本章代码基于 LangChain 0.3.x 编写。如果你曾使用过 0.1/0.2 版本，注意以下变化：

• langchain 核心精简，功能迁移到 langchain-core
• 旧的 ConversationBufferMemory 等 Memory 类标记为 legacy，推荐使用 Runnable + MessageHistory 架构
• Agent 推荐使用 create_tool_calling_agent（工具调用型 Agent）替代旧版 ReAct Agent

Chain 链式调用

Chain 是 LangChain 的核心抽象，表示"一系列有序操作"。就像前端的数据流管道，数据从一端流入，经过多个处理步骤，从另一端流出。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 初始化模型（兼容 OpenAI 格式的 API 都可以）
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="sk-your-key",
    base_url="https://api.openai.com/v1"
)

# 构建 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，用{style}风格回答问题。"),
    ("human", "{question}")
])

# 输出解析器：将模型输出转为字符串
output_parser = StrOutputParser()

# 组装 Chain：prompt -> llm -> parser
chain = prompt | llm | output_parser

# 调用 Chain（像调用函数一样）
result = chain.invoke({
    "role": "前端专家",
    "style": "幽默",
    "question": "为什么 CSS 居中这么难？"
})
print(result)

# 批量处理
questions = [
    {"role": "前端专家", "style": "简洁", "question": "什么是闭包？"},
    {"role": "后端专家", "style": "详细", "question": "什么是事务？"}
]
results = chain.batch(questions)
for r in results:
    print(r)

PromptTemplate

LangChain 提供了多种 Prompt 模板，让提示词管理更加工程化：

from langchain_core.prompts import (
    PromptTemplate,
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)

# 1. 简单字符串模板
prompt = PromptTemplate.from_template("""
将以下文本翻译成 {language}：
{text}
""")

# 2. 聊天消息模板（推荐）
chat_prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template(
        "你是一位资深 {domain} 工程师。"
    ),
    MessagesPlaceholder(variable_name="history"),  # 插入历史消息
    HumanMessagePromptTemplate.from_template("{input}")
])

# 3. 部分填充模板（复用）
code_reviewer_prompt = PromptTemplate.from_template("""
请审查以下 {language} 代码，关注：
1. 潜在 Bug
2. 性能问题
3. 可读性

代码：
```{language}
{code}

""")

创建一个专门审查 Python 的模板

python_reviewer = code_reviewer_prompt.partial(language="python")

### Memory 记忆（对话历史管理）

LangChain 0.3+ 推荐使用 **Runnable + MessageHistory** 架构管理对话历史，替代旧的 `ConversationBufferMemory` 等 legacy API。新架构更清晰、更可组合：

```python
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 存储会话历史的字典（生产环境可用 Redis、数据库替代）
store = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    """根据会话 ID 获取或创建消息历史."""
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 构建带历史记录的 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful 的助手，记住用户告诉你的信息。"),
    MessagesPlaceholder(variable_name="history"),  # 历史消息插入位置
    ("human", "{input}")
])

# 组装 Chain
chain = prompt | llm | StrOutputParser()

# 包装为带历史记忆的 Chain
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 使用：指定 session_id 即可保持对话上下文
config = {"configurable": {"session_id": "user_001"}}

response1 = chain_with_history.invoke(
    {"input": "我叫小明，我喜欢 Python。"},
    config=config
)
print(response1)

# 同一 session_id，模型能记住"小明"
response2 = chain_with_history.invoke(
    {"input": "我刚才说了我叫什么？"},
    config=config
)
print(response2)

新旧对比

特性	Legacy Memory (0.1/0.2)	新架构 (0.3+)
核心类	ConversationBufferMemory	InMemoryChatMessageHistory + RunnableWithMessageHistory
集成方式	通过 ConversationChain 隐式集成	通过 RunnableWithMessageHistory 显式包装
存储扩展	需实现 BaseMemory 子类	任何返回 BaseChatMessageHistory 的函数均可
灵活性	固定模式	可自定义历史压缩、过滤、摘要策略

Agents 与 Tools

LangChain 的 Agent 可以自动分析用户意图，选择合适的工具执行。LangChain 0.3+ 推荐利用 LLM 的原生 Tool Calling 能力（OpenAI、Claude、Qwen 等主流模型均支持），相比传统的 ReAct 文本解析方式更稳定、更可靠。

方式一：Tool Calling Agent（推荐）

利用 LLM 的原生工具调用能力，模型直接输出工具名和参数 JSON，无需复杂的文本解析：

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 1. 定义工具（用 @tool 装饰器自动提取函数签名和文档）
@tool
def search_web(query: str) -> str:
    """搜索网络获取实时信息。当用户询问最新事件、新闻或你不知道的信息时使用。"""
    # 实际项目中接入搜索引擎 API
    return f"【模拟搜索结果】关于 '{query}' 的最新信息：..."

@tool
def calculate(expression: str) -> str:
    """执行数学计算。当用户需要进行数值计算时使用。"""
    import ast
    import operator

    allowed_ops = {
        ast.Add: operator.add, ast.Sub: operator.sub,
        ast.Mult: operator.mul, ast.Div: operator.truediv,
        ast.Pow: operator.pow, ast.USub: operator.neg,
        ast.Mod: operator.mod, ast.FloorDiv: operator.floordiv,
    }

    def safe_eval(node):
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        elif isinstance(node, ast.BinOp):
            op = allowed_ops.get(type(node.op))
            if not op: raise ValueError(f"不支持的运算符")
            return op(safe_eval(node.left), safe_eval(node.right))
        elif isinstance(node, ast.UnaryOp):
            op = allowed_ops.get(type(node.op))
            if not op: raise ValueError(f"不支持的一元运算符")
            return op(safe_eval(node.operand))
        elif isinstance(node, ast.Expression):
            return safe_eval(node.body)
        raise ValueError("不支持的表达式")

    try:
        result = safe_eval(ast.parse(expression, mode="eval"))
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

@tool
def get_current_time() -> str:
    """获取当前时间。"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 2. 工具列表
tools = [search_web, calculate, get_current_time]

# 3. 构建 Prompt（Tool Calling Agent 需要包含工具占位符）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful 的 AI 助手，可以使用工具帮助用户解决问题。"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 工具执行历史
])

# 4. 创建 Agent（利用 LLM 原生 Tool Calling）
agent = create_tool_calling_agent(llm, tools, prompt)

# 5. 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印思考过程
    handle_parsing_errors=True
)

# 6. 运行
result = agent_executor.invoke({
    "input": "现在几点了？123 乘以 456 等于多少？"
})
print(result["output"])

方式二：ReAct Agent（传统方式，兼容性好）

如果使用的模型不支持原生 Tool Calling（如部分开源模型），可以使用 ReAct Agent，它通过文本推理链（Thought → Action → Observation）驱动工具调用：

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

# 加载 ReAct Prompt 模板（带本地 fallback）
try:
    prompt = hub.pull("hwchase17/react")
except Exception:
    from langchain_core.prompts import PromptTemplate
    prompt = PromptTemplate.from_template("""Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}""")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

两种方式的选择

特性	Tool Calling Agent	ReAct Agent
稳定性	高（结构化 JSON 输出）	中（依赖文本解析）
模型兼容性	需模型支持 tool calling	几乎所有 LLM
推荐场景	生产环境、OpenAI/Claude/Qwen	开源小模型、教学演示
速度	更快（少一轮推理）	略慢

7.3 LangChain 实战

构建一个多功能 Agent，能查天气（模拟）、算数学、写代码。

#!/usr/bin/env python3
"""
multi_tool_agent.py
多功能 Agent：天气查询 + 数学计算 + 代码生成
基于 LangChain 0.3+ Tool Calling Agent
"""

import os
import ast
import operator
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from dotenv import load_dotenv

load_dotenv()

# 初始化 LLM
llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    temperature=0.3
)

# ========== 工具定义 ==========

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。当用户询问天气时使用。"""
    weather_data = {
        "北京": {"temp": 22, "condition": "晴", "humidity": 45},
        "上海": {"temp": 26, "condition": "多云", "humidity": 65},
        "深圳": {"temp": 30, "condition": "小雨", "humidity": 80},
        "杭州": {"temp": 24, "condition": "阴", "humidity": 55}
    }
    data = weather_data.get(city)
    if not data:
        return f"暂无 {city} 的天气数据，试试：北京、上海、深圳、杭州"
    return f"{city}今天{data['condition']}，气温 {data['temp']}°C，湿度 {data['humidity']}%"

@tool
def calculate(expression: str) -> str:
    """执行数学计算。当用户需要进行加减乘除或复杂计算时使用。"""
    allowed_ops = {
        ast.Add: operator.add, ast.Sub: operator.sub,
        ast.Mult: operator.mul, ast.Div: operator.truediv,
        ast.Pow: operator.pow, ast.USub: operator.neg,
        ast.Mod: operator.mod, ast.FloorDiv: operator.floordiv,
    }

    def safe_eval(node):
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        elif isinstance(node, ast.BinOp):
            op = allowed_ops.get(type(node.op))
            if not op: raise ValueError("不支持的运算符")
            return op(safe_eval(node.left), safe_eval(node.right))
        elif isinstance(node, ast.UnaryOp):
            op = allowed_ops.get(type(node.op))
            if not op: raise ValueError("不支持的一元运算符")
            return op(safe_eval(node.operand))
        elif isinstance(node, ast.Expression):
            return safe_eval(node.body)
        raise ValueError("不支持的表达式")

    try:
        result = safe_eval(ast.parse(expression, mode="eval"))
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

@tool
def generate_code(description: str, language: str = "python") -> str:
    """生成代码示例。当用户要求写代码或需要代码示例时使用。"""
    return f"【代码生成器】将为您生成 {language} 代码：{description}"

tools = [get_weather, calculate, generate_code]

# ========== Agent 配置（Tool Calling 方式）==========

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful 的 AI 助手，可以使用工具帮助用户解决问题。"),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# 使用 create_tool_calling_agent（LangChain 0.3+ 推荐）
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True
)

# ========== 交互循环 ==========

def main():
    print("=" * 50)
    print("  多功能 AI Agent")
    print("  支持：天气查询 / 数学计算 / 代码生成")
    print("  输入 /quit 退出")
    print("=" * 50)
    print()

    while True:
        user_input = input("你: ").strip()
        if user_input == "/quit":
            break
        if not user_input:
            continue

        try:
            result = agent_executor.invoke({"input": user_input})
            print(f"\n助手: {result['output']}\n")
        except Exception as e:
            print(f"\n出错了: {e}\n")

if __name__ == "__main__":
    main()

7.4 AutoGen 多智能体

AutoGen 是微软开源的多智能体框架，核心思想是"对话式编程"——让多个 AI Agent 像团队成员一样协作完成任务。

前端类比

AutoGen 就像一个敏捷开发团队：UserProxyAgent 是产品经理（提需求），AssistantAgent 是程序员（写代码），他们之间通过"对话"（消息传递）协作完成项目。

核心概念

• ConversableAgent：所有 Agent 的基类，能发送和接收消息
• UserProxyAgent：代表人类用户，可以执行代码、调用工具
• AssistantAgent：AI 助手，负责推理和生成回复
• GroupChat：多 Agent 群聊，自动管理发言顺序

pip install pyautogen

import os
from autogen import ConversableAgent, UserProxyAgent

# 配置 LLM
config_list = [{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
}]

# 创建助手 Agent
assistant = ConversableAgent(
    name="代码助手",
    system_message="你是一位 Python 专家。编写简洁、可运行的代码，并解释关键部分。",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER"
)

# 创建用户代理 Agent
user_proxy = UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False  # 本地执行，不用 Docker
    }
)

# 启动对话：用户代理向助手发送任务
user_proxy.initiate_chat(
    assistant,
    message="写一个 Python 脚本，计算斐波那契数列前 20 项，并保存到 fibonacci.txt"
)

# 输出：
# 助手会生成代码，用户代理会自动执行并返回结果
# 如果代码有错误，助手会修复，形成自动迭代

多 Agent 协作

from autogen import GroupChat, GroupChatManager

# 定义多个专家 Agent
coder = ConversableAgent(
    name="程序员",
    system_message="你负责编写代码。只输出代码，不输出解释。",
    llm_config={"config_list": config_list}
)

reviewer = ConversableAgent(
    name="代码审查员",
    system_message="你负责审查代码质量。检查 Bug、性能和可读性，提出改进建议。",
    llm_config={"config_list": config_list}
)

tester = ConversableAgent(
    name="测试员",
    system_message="你负责编写测试用例。确保代码覆盖边界情况。",
    llm_config={"config_list": config_list}
)

# 创建群聊
group_chat = GroupChat(
    agents=[user_proxy, coder, reviewer, tester],
    messages=[],
    max_round=12  # 最多 12 轮对话
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": config_list}
)

# 启动协作
user_proxy.initiate_chat(
    manager,
    message="开发一个函数，验证邮箱地址格式，需要包含单元测试。"
)

7.5 MCP (Model Context Protocol)

MCP 是 Anthropic 于 2024 年底推出的开放协议，旨在标准化 LLM 与外部工具、数据源之间的通信方式。

为什么 MCP 重要

在 MCP 之前，每个框架（LangChain、AutoGen、LlamaIndex）都有自己的工具定义格式。开发者需要为不同框架重复编写工具适配代码。MCP 就像"AI 领域的 USB-C"——统一接口，一次编写，到处使用。

前端类比

MCP 就像 REST API 标准化了前后端通信。以前每个后端框架有自己的 RPC 协议，REST 出现后，前端可以用统一的方式（HTTP + JSON）调用任何后端服务。

MCP Server 与 Client 架构

• MCP Server：暴露工具和数据源的服务端。例如：文件系统 Server、数据库 Server、GitHub Server
• MCP Client：消费 Server 能力的客户端。例如：Claude Desktop、Cursor、自定义 Agent

# MCP Server 示例（使用官方 Python SDK）
# pip install mcp

from mcp.server import Server
from mcp.types import TextContent
import mcp.server.stdio

server = Server("file-reader")

@server.list_tools()
async def list_tools():
    return [{
        "name": "read_file",
        "description": "读取文件内容",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "文件路径"}
            },
            "required": ["path"]
        }
    }]

@server.call_tool()
async def call_tool(name, arguments):
    if name == "read_file":
        with open(arguments["path"], "r") as f:
            content = f.read()
        return [TextContent(type="text", text=content)]

# 启动 Server
if __name__ == "__main__":
    mcp.server.stdio.run_server(server)

MCP Client 完整使用示例

Server 定义了工具，Client 则负责连接 Server 并调用工具。以下是用 Python 编写 MCP Client 的完整示例：

"""MCP Client 示例 - 连接本地 Server 并调用工具.

安装: pip install mcp
"""

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def use_mcp_server():
    """启动并连接本地 MCP Server，调用其工具."""

    # 配置 Server 启动参数（通过 stdio 通信）
    server_params = StdioServerParameters(
        command="python",
        args=["my_mcp_server.py"],  # 你的 Server 脚本
        env=None
    )

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化
            await session.initialize()

            # 获取 Server 提供的工具列表
            tools = await session.list_tools()
            print(f"可用工具: {[tool.name for tool in tools.tools]}")

            # 调用工具
            result = await session.call_tool(
                "read_file",
                arguments={"path": "./data/report.txt"}
            )
            print(f"工具返回: {result}")

# asyncio.run(use_mcp_server())

常用官方 MCP Server 生态

MCP 生态正在快速扩展，以下是已可用的官方和社区 Server：

Server	功能	RPA 场景
filesystem	读写本地文件、遍历目录	读取配置文件、保存处理结果
sqlite	查询本地 SQLite 数据库	RPA 数据记录查询
fetch	获取网页内容	爬取信息、监控网页变化
github	读取仓库、Issue、PR	自动化的代码审查辅助
slack	发送和读取 Slack 消息	任务完成通知
puppeteer	浏览器自动化	网页填报、截图存档

安装官方 Server（以 filesystem 为例）：

# 使用 npx 安装并运行（要求 Node.js）
npx -y @modelcontextprotocol/server-filesystem /path/to/allowed/dir

# 或在 Claude Desktop 配置中添加：
# ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
# %APPDATA%\Claude\claude_desktop_config.json (Windows)

RPA 场景：自定义 Excel MCP Server

为 RPA 场景编写一个自定义 MCP Server，让任何 MCP Client（如 Claude Desktop、Cursor）都能操作 Excel 文件：

"""Excel MCP Server - 让 AI 通过 MCP 协议操作 Excel 文件.

安装: pip install mcp openpyxl
"""

import json
from mcp.server import Server
from mcp.types import TextContent
import mcp.server.stdio
from openpyxl import load_workbook
from pathlib import Path

server = Server("excel-operator")

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "read_excel_sheet",
            "description": "读取 Excel 文件的指定工作表内容",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Excel 文件路径"},
                    "sheet_name": {"type": "string", "description": "工作表名称，默认第一个"},
                    "max_rows": {"type": "integer", "description": "最大读取行数，默认 100"}
                },
                "required": ["file_path"]
            }
        },
        {
            "name": "write_excel_cell",
            "description": "向 Excel 单元格写入数据",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Excel 文件路径"},
                    "sheet_name": {"type": "string", "description": "工作表名称"},
                    "cell": {"type": "string", "description": "单元格地址，如 A1"},
                    "value": {"type": "string", "description": "要写入的值"}
                },
                "required": ["file_path", "cell", "value"]
            }
        },
        {
            "name": "list_excel_sheets",
            "description": "列出 Excel 文件中的所有工作表",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Excel 文件路径"}
                },
                "required": ["file_path"]
            }
        }
    ]

@server.call_tool()
async def call_tool(name, arguments):
    file_path = arguments.get("file_path")

    if not Path(file_path).exists():
        return [TextContent(type="text", text=f"错误: 文件不存在 {file_path}")]

    if name == "read_excel_sheet":
        wb = load_workbook(file_path, data_only=True)
        sheet_name = arguments.get("sheet_name") or wb.sheetnames[0]
        ws = wb[sheet_name]
        max_rows = arguments.get("max_rows", 100)

        data = []
        for row in ws.iter_rows(min_row=1, max_row=max_rows, values_only=True):
            data.append([str(cell) if cell is not None else "" for cell in row])

        return [TextContent(type="text", text=json.dumps(data, ensure_ascii=False, indent=2))]

    elif name == "write_excel_cell":
        wb = load_workbook(file_path)
        sheet_name = arguments.get("sheet_name") or wb.sheetnames[0]
        ws = wb[sheet_name]
        ws[arguments["cell"]] = arguments["value"]
        wb.save(file_path)
        return [TextContent(type="text", text=f"已写入 {arguments['cell']}: {arguments['value']}")]

    elif name == "list_excel_sheets":
        wb = load_workbook(file_path)
        return [TextContent(type="text", text=json.dumps(wb.sheetnames))]

    return [TextContent(type="text", text="未知工具")]

if __name__ == "__main__":
    mcp.server.stdio.run_server(server)

MCP 发展趋势

• 标准化：MCP 正在迅速成为 LLM 工具调用的"事实标准"
• 生态扩展：文件系统、数据库、浏览器、GitHub、Slack 等官方 Server 持续增加
• IDE 集成：Claude Desktop、Cursor、Windsurf 等已原生支持 MCP
• RPA 机遇：用 MCP 封装企业内部的 RPA 能力（操作 ERP、CRM、Excel），让 AI 统一调度

建议关注官方仓库 github.com/modelcontextprotocol 获取最新动态。

7.6 RAG 基础

RAG（Retrieval-Augmented Generation，检索增强生成）是解决 LLM "幻觉"和"知识过时"问题的核心技术。

为什么需要 RAG

LLM 有两个固有缺陷：

1. 幻觉：模型会"编造"看似合理但实际错误的信息
2. 知识截止：训练数据有截止日期，无法获取最新信息

RAG 的思路是：在生成回答前，先从外部知识库检索相关信息，把检索结果作为上下文提供给模型，让模型"有据可依"。

前端类比

RAG 就像前端的数据获取流程：组件渲染前，先调用 API 获取数据（检索），再把数据传入组件渲染（生成）。没有数据获取的组件只能显示静态内容（类似 LLM 的预训练知识）。

Embedding 与向量

Embedding 是将文本转换为数值向量的技术。语义相近的文本，其向量在空间中距离也更近。

from langchain_openai import OpenAIEmbeddings

# 初始化 Embedding 模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key="sk-your-key"
)

# 将文本转为向量
texts = [
    "Python 是一种解释型编程语言",
    "JavaScript 是前端开发的核心语言",
    "Python 适合数据分析和人工智能"
]

vectors = embeddings.embed_documents(texts)
print(f"文本数量: {len(vectors)}")
print(f"每个向量维度: {len(vectors[0])}")
# 输出: 文本数量: 3, 每个向量维度: 1536

# 计算相似度（余弦相似度）
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# "Python 数据分析" 与第一个文本更相关
query = embeddings.embed_query("Python 数据分析")
sim1 = cosine_similarity(query, vectors[0])  # ~0.85
sim2 = cosine_similarity(query, vectors[1])  # ~0.45
print(f"与'Python 语言'相似度: {sim1:.3f}")
print(f"与'JavaScript'相似度: {sim2:.3f}")

文本分割

长文档需要分割成小块（Chunk）才能存入向量数据库。分割策略直接影响检索质量：

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter
)

# 1. 递归字符分割器（推荐）
# 按优先级尝试分割：段落 -> 句子 -> 单词 -> 字符
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每块最大 500 字符
    chunk_overlap=50,    # 块之间重叠 50 字符（保持上下文连贯）
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)

# 2. 固定字符分割器
char_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200
)

# 使用示例
long_text = """Python 是由 Guido van Rossum 于 1991 年创建的...
（此处省略一篇长文章）"""

chunks = recursive_splitter.split_text(long_text)
print(f"分割为 {len(chunks)} 个块")
for i, chunk in enumerate(chunks[:3]):
    print(f"块 {i}: {chunk[:50]}...")

ChromaDB 向量数据库

ChromaDB 是轻量级的开源向量数据库，适合入门和中小型项目。

pip install chromadb langchain-chroma

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# 初始化 Embedding 和向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建/加载向量数据库
vectorstore = Chroma(
    collection_name="python_tutorial",
    embedding_function=embeddings,
    persist_directory="./chroma_db"  # 数据持久化目录
)

# 添加文档
from langchain_core.documents import Document

docs = [
    Document(page_content="Python 的列表推导式是一种简洁创建列表的方式。", metadata={"source": "ch2"}),
    Document(page_content="装饰器是 Python 中修改函数行为的高级特性。", metadata={"source": "ch3"}),
    Document(page_content="asyncio 是 Python 的异步编程库。", metadata={"source": "ch4"})
]

vectorstore.add_documents(docs)

# 相似性搜索
results = vectorstore.similarity_search("如何创建列表？", k=2)
for doc in results:
    print(f"内容: {doc.page_content}")
    print(f"来源: {doc.metadata['source']}")
    print("---")

# 带分数的搜索（分数越低越相似）
results_with_scores = vectorstore.similarity_search_with_score("异步编程", k=2)
for doc, score in results_with_scores:
    print(f"分数: {score:.4f}, 内容: {doc.page_content}")

RAG 完整检索流程

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 相似度搜索
    search_kwargs={"k": 3}     # 返回前 3 个结果
)

# 创建 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 组装 RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 将所有检索结果"塞"进 Prompt
    retriever=retriever,
    return_source_documents=True,  # 返回来源文档
    verbose=True
)

# 提问
result = qa_chain.invoke({"query": "Python 装饰器是什么？"})
print(f"回答: {result['result']}")
print("\n参考来源:")
for doc in result["source_documents"]:
    print(f"  - {doc.page_content[:60]}...")

7.7 实战：个人知识库问答系统

综合运用 LangChain + ChromaDB，构建一个能读取本地文档并回答问题的知识库系统。

#!/usr/bin/env python3
"""
knowledge_base_qa.py
基于 LangChain + ChromaDB 的个人知识库问答系统

功能：
1. 加载本地文档（txt, md, pdf）
2. 自动分割、Embedding、存入向量库
3. 支持自然语言查询
"""

import os
import sys
from pathlib import Path
from dotenv import load_dotenv

from langchain_community.document_loaders import (
    TextLoader,
    UnstructuredMarkdownLoader,
    DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

load_dotenv()

class KnowledgeBaseQA:
    def __init__(self, docs_dir="./docs", db_dir="./chroma_db"):
        self.docs_dir = Path(docs_dir)
        self.db_dir = Path(db_dir)

        # 初始化 Embedding
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL")
        )

        # 初始化 LLM
        self.llm = ChatOpenAI(
            model=os.getenv("LLM_MODEL", "gpt-4o-mini"),
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL"),
            temperature=0.3
        )

        self.vectorstore = None
        self.qa_chain = None

    def load_documents(self):
        """加载文档目录中的所有支持文件"""
        print(f"正在加载文档: {self.docs_dir}")

        documents = []

        # 加载 txt 文件
        if list(self.docs_dir.glob("*.txt")):
            txt_loader = DirectoryLoader(
                str(self.docs_dir),
                glob="*.txt",
                loader_cls=TextLoader,
                loader_kwargs={"encoding": "utf-8"}
            )
            documents.extend(txt_loader.load())

        # 加载 md 文件
        if list(self.docs_dir.glob("*.md")):
            md_loader = DirectoryLoader(
                str(self.docs_dir),
                glob="*.md",
                loader_cls=UnstructuredMarkdownLoader
            )
            documents.extend(md_loader.load())

        print(f"共加载 {len(documents)} 个文档")
        return documents

    def build_knowledge_base(self):
        """构建知识库：分割 -> Embedding -> 存储"""
        documents = self.load_documents()

        if not documents:
            print("警告: 没有找到任何文档")
            return False

        # 文本分割
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""]
        )
        chunks = splitter.split_documents(documents)
        print(f"分割为 {len(chunks)} 个文本块")

        # 存入向量数据库
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=str(self.db_dir)
        )
        print(f"知识库已保存到: {self.db_dir}")
        return True

    def load_existing_kb(self):
        """加载已存在的知识库"""
        if not self.db_dir.exists():
            return False

        self.vectorstore = Chroma(
            persist_directory=str(self.db_dir),
            embedding_function=self.embeddings
        )
        print("已加载现有知识库")
        return True

    def create_qa_chain(self):
        """创建问答链"""
        if not self.vectorstore:
            raise ValueError("知识库未初始化")

        # 对话记忆
        memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )

        # 检索器
        retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 4}
        )

        # 对话式 RAG 链
        self.qa_chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=retriever,
            memory=memory,
            return_source_documents=True,
            verbose=False
        )

    def ask(self, question):
        """提问"""
        if not self.qa_chain:
            raise ValueError("问答链未初始化")

        result = self.qa_chain.invoke({"question": question})
        return {
            "answer": result["answer"],
            "sources": [doc.page_content[:100] + "..."
                       for doc in result.get("source_documents", [])]
        }

    def run_interactive(self):
        """交互式问答"""
        print("\n" + "=" * 50)
        print("  个人知识库问答系统")
        print("  输入 /quit 退出，/rebuild 重建知识库")
        print("=" * 50 + "\n")

        # 尝试加载或构建知识库
        if not self.load_existing_kb():
            if self.docs_dir.exists():
                self.build_knowledge_base()
            else:
                print(f"请先在 {self.docs_dir} 目录放入文档")
                return

        self.create_qa_chain()

        while True:
            question = input("问题: ").strip()
            if question == "/quit":
                break
            if question == "/rebuild":
                self.build_knowledge_base()
                self.create_qa_chain()
                continue
            if not question:
                continue

            try:
                result = self.ask(question)
                print(f"\n回答: {result['answer']}\n")
                if result["sources"]:
                    print("参考片段:")
                    for i, src in enumerate(result["sources"][:2], 1):
                        print(f"  {i}. {src}")
                print()
            except Exception as e:
                print(f"错误: {e}\n")

def main():
    # 创建示例文档目录
    docs_dir = Path("./docs")
    docs_dir.mkdir(exist_ok=True)

    # 创建示例文档
    sample_file = docs_dir / "python_notes.txt"
    if not sample_file.exists():
        sample_file.write_text("""Python 学习笔记

列表推导式：
列表推导式是 Python 中创建列表的简洁方式。
语法：[expression for item in iterable if condition]
示例：squares = [x**2 for x in range(10)]

装饰器：
装饰器是一种修改函数行为的高级特性。
使用 @ 语法糖，本质上是一个接收函数并返回函数的函数。
常用于日志记录、权限检查、缓存等场景。

虚拟环境：
Python 使用 venv 模块创建隔离的开发环境。
命令：python -m venv myenv
激活：source myenv/bin/activate (Linux/Mac)
      myenv\\Scripts\\activate (Windows)
""", encoding="utf-8")
        print(f"已创建示例文档: {sample_file}")

    # 启动系统
    kb = KnowledgeBaseQA(docs_dir="./docs", db_dir="./chroma_db")
    kb.run_interactive()

if __name__ == "__main__":
    main()

扩展方向

在此基础上，你可以添加：多格式支持（PDF、Word）、增量更新（只处理新文件）、Web 界面（Gradio/Streamlit）、混合检索（关键词 + 向量）、重排序（Rerank）等高级功能。

7.8 LangGraph —— 状态机工作流

LangGraph 是 LangChain 团队推出的库，用于构建有状态的、多角色的 Agent 工作流。 如果说 LangChain 的 Chain 是"管道"（数据单向流动），LangGraph 就是"状态机"（数据可以在节点间循环、分支）。 这非常像前端的 Redux 或 XState 的概念——中心化的状态 + 状态转换图。

前端类比

LangChain Chain = 数据管道（pipe/waterfall）；LangGraph = 状态机（Redux store + reducer）。 当你的 Agent 需要在不同策略间切换、循环执行直到满足条件时，LangGraph 是更好的选择。

"""LangGraph 示例 —— 带条件判断的 Agent 工作流 (pip install langgraph)"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

# 定义状态（类似 Redux store 的 state 类型）
class AgentState(TypedDict):
    messages: list[str]
    next_action: str
    result: str | None

# 定义节点（每个节点是一个处理步骤）
def analyze_task(state: AgentState) -> AgentState:
    """分析任务，决定下一步"""
    last_msg = state["messages"][-1] if state["messages"] else ""
    if "计算" in last_msg:
        return {**state, "next_action": "calculate"}
    elif "翻译" in last_msg:
        return {**state, "next_action": "translate"}
    else:
        return {**state, "next_action": "answer"}

def calculate(state: AgentState) -> AgentState:
    """执行计算"""
    state["result"] = "计算结果: 42"
    return state

def translate(state: AgentState) -> AgentState:
    """执行翻译"""
    state["result"] = "翻译结果: Hello World"
    return state

def direct_answer(state: AgentState) -> AgentState:
    """直接回答"""
    state["result"] = "这是一个通用问题的回答"
    return state

# 路由函数：根据状态决定下一步
def router(state: AgentState) -> str:
    return state["next_action"]

# 构建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("analyze", analyze_task)
graph.add_node("calculate", calculate)
graph.add_node("translate", translate)
graph.add_node("direct_answer", direct_answer)

# 添加边
graph.set_entry_point("analyze")  # 入口
graph.add_conditional_edges("analyze", router, {
    "calculate": "calculate",
    "translate": "translate",
    "answer": "direct_answer"
})
graph.add_edge("calculate", END)   # 终止
graph.add_edge("translate", END)
graph.add_edge("direct_answer", END)

# 编译并运行
app = graph.compile()
result = app.invoke({"messages": ["帮我计算 2+2"], "next_action": "", "result": None})
print(result["result"])  # 计算结果: 42

7.9 LlamaIndex 简介

LlamaIndex 是与 LangChain 互补的框架，专注于数据索引和检索。 如果说 LangChain 是"AI 应用的 React"，LlamaIndex 就是"AI 数据的数据库引擎"。 它在 RAG（检索增强生成）场景中特别强大。

"""LlamaIndex 示例 —— 快速构建 RAG 应用 (pip install llama-index)"""
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# 1. 加载文档（支持 PDF、Word、Markdown 等格式）
documents = SimpleDirectoryReader("./docs").load_data()

# 2. 自动分块、Embedding、索引（一行代码！）
index = VectorStoreIndex.from_documents(documents)

# 3. 创建查询引擎
query_engine = index.as_query_engine(
    similarity_top_k=3,  # 检索前 3 个最相关的文档块
    streaming=False
)

# 4. 提问
response = query_engine.query("这个项目使用什么数据库？")
print(response)

# LlamaIndex 的优势：
# - 内置多种文档加载器（PDF、Word、Notion、GitHub 等）
# - 多种检索策略（关键词 + 向量混合检索）
# - 支持结构化数据查询（结合 SQL）
# - 内置评估框架

LangChain vs LlamaIndex 选择建议

场景	推荐
构建对话 Agent（多工具、多步骤）	LangChain + LangGraph
文档问答 / 知识库（核心是检索）	LlamaIndex
复杂 AI 应用（两者都用到）	两者结合使用
快速原型（RAG 应用）	LlamaIndex（更简单）

7.10 Agent 评估方法

Agent 开发的最大挑战之一是评估——你怎么知道 Agent 的回复质量是好是坏？ 以下是一些实用的评估策略：

LLM-as-Judge 评估

用另一个 LLM 来评估 Agent 的输出，就像代码审查一样：

"""LLM-as-Judge 评估示例"""
def evaluate_response(question, answer, expected_keywords, model="gpt-4o-mini"):
    """用 LLM 评估 Agent 回答质量"""

    evaluation_prompt = f"""请评估以下 AI 助手的回答质量：

用户问题: {question}
AI 回答: {answer}
期望包含的关键信息: {", ".join(expected_keywords)}

请按以下维度打分（1-5）：
1. 准确性：回答是否包含正确的关键信息？
2. 完整性：回答是否覆盖了用户的全部问题？
3. 清晰度：回答是否易于理解？
4. 安全性：回答是否有潜在的安全风险？

以 JSON 格式输出：
{{"accuracy": int, "completeness": int, "clarity": int, "safety": int, "overall": float, "comments": str}}"""

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": evaluation_prompt}],
        response_format={"type": "json_object"}
    )

    return json.loads(response.choices[0].message.content)

# 构建测试用例集
test_cases = [
    {"question": "如何安全地解析用户输入？", "expected_keywords": ["验证", "sanitize", "SQL注入", "XSS"]},
    {"question": "什么是列表推导式？", "expected_keywords": ["列表", "推导式", "简洁", "for", "if"]},
]

for case in test_cases:
    answer = your_agent.ask(case["question"])
    score = evaluate_response(case["question"], answer, case["expected_keywords"])
    print(f"Q: {case['question'][:30]}...")
    print(f"Score: {score['overall']}/5")
    print(f"Comments: {score['comments']}")
    print("---")

7.11 CrewAI —— 多智能体协作框架

CrewAI 是 2024 年最受欢迎的 multi-agent 协作框架之一。与 LangChain 的单 Agent 编排不同，CrewAI 强调"团队"概念：多个 Agent 各自扮演不同角色（研究员、写手、编辑），通过**流程（Process）**协作完成复杂任务。

安装

pip install crewai

核心概念

概念	说明	前端类比
Agent	智能体，有角色、目标、背景故事	组件/微服务
Task	分配给 Agent 的具体任务	函数/Job
Crew	Agent + Task 的编排组合	项目/团队
Process	协作模式：串行(sequential) 或 层级(hierarchical)	工作流引擎

实战：自动化内容创作团队

假设你需要让 AI 自动完成一篇技术博客：研究员收集资料 → 写手撰写文章 → 编辑审核润色。

"""CrewAI 实战 - 自动化内容创作团队.

角色分工：
  1. 研究员：搜索和整理技术资料
  2. 写手：根据资料撰写文章
  3. 编辑：审核质量、优化表达
"""

import os
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# 初始化 LLM（CrewAI 兼容 LangChain 的 LLM）
llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
)

# ========== 定义 Agent（团队成员）==========

researcher = Agent(
    role="技术研究员",
    goal="收集最新、最准确的技术资料，为写作提供素材",
    backstory="你是一位资深技术编辑，擅长快速理解和整理复杂的技术概念。你总是提供结构清晰、引用准确的资料。",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="技术写手",
    goal="将技术资料转化为通俗易懂的中文技术文章",
    backstory="你是一位拥有 10 年经验的技术博主，擅长用生动的比喻和代码示例解释复杂概念。你的文章总是让读者拍案叫绝。",
    llm=llm,
    verbose=True
)

editor = Agent(
    role="内容编辑",
    goal="确保文章质量，修正错误，优化表达",
    backstory="你是一位挑剔但公正的内容编辑，对技术准确性要求极高。你会指出任何不准确或表达不清的地方。",
    llm=llm,
    verbose=True
)

# ========== 定义任务 ==========

task_research = Task(
    description="研究主题：'Python 异步编程中的 asyncio'。\n"
                "收集以下信息：\n"
                "1. asyncio 的核心概念（事件循环、协程、任务）\n"
                "2. 与多线程/多进程的对比\n"
                "3. 常见的使用场景和最佳实践\n"
                "4. 一个完整的代码示例\n"
                "输出为结构化的研究笔记。",
    expected_output="一份结构化的研究笔记，包含准确的技术概念、对比表格和代码示例",
    agent=researcher
)

task_write = Task(
    description="基于研究员提供的资料，撰写一篇面向前端开发者的 Python asyncio 入门教程。\n"
                "要求：\n"
                "1. 用前端开发者熟悉的概念做类比（如 Promise、Event Loop）\n"
                "2. 包含可运行的代码示例\n"
                "3. 文章长度约 1500 字\n"
                "4. 语气友好、专业",
    expected_output="一篇完整的 Markdown 格式技术文章",
    agent=writer
)

task_edit = Task(
    description="审核写手提交的文章，检查：\n"
                "1. 技术准确性：代码是否能正常运行？概念解释是否正确？\n"
                "2. 表达清晰度：是否有晦涩难懂的句子？\n"
                "3. 结构完整性：开头、正文、结尾是否完整？\n"
                "直接输出修改后的最终版本。",
    expected_output="编辑后的最终文章版本，附带修改说明",
    agent=editor
)

# ========== 组建团队并执行 ==========

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[task_research, task_write, task_edit],
    process=Process.sequential,  # 串行执行：研究 → 写作 → 编辑
    verbose=True
)

result = crew.kickoff()
print("=" * 50)
print("最终输出:")
print(result)

CrewAI 适用场景

• 内容创作：研究 → 写作 → 编辑的流水线
• 代码审查：程序员写代码 → 审查员检查 → 测试员写用例
• 数据分析：数据工程师清洗 → 分析师建模 → 业务经理解读
• RPA 复杂流程：多步骤、多角色的自动化任务

7.12 PydanticAI —— 类型安全的 Agent 框架

PydanticAI 是由 Pydantic 团队开发的 Agent 框架，核心理念是用类型系统驱动 Agent 开发。如果你熟悉 TypeScript 的类型安全，会立刻爱上它的设计哲学。

安装

pip install pydantic-ai

核心特点

1. 严格的类型安全：工具参数、模型输出全部通过 Pydantic 模型校验
2. 依赖注入系统：通过 deps_type 注入配置、数据库连接等依赖
3. 原生异步支持：基于 asyncio，性能优异
4. 多模型支持：OpenAI、Anthropic、Gemini、Ollama 等统一接口

实战：带类型安全的天气 Agent

"""PydanticAI 实战 - 类型安全的 Agent.

特点：
  - 工具参数和返回值都有类型注解
  - 依赖注入（deps）传递配置
  - 结构化输出（result_type）
"""

import os
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

# ========== 定义结构化输出模型 ==========

class WeatherResult(BaseModel):
    """天气查询结果的结构化输出."""
    city: str
    temperature: int
    condition: str
    suggestion: str

# ========== 定义依赖（配置）==========

class WeatherDeps(BaseModel):
    """Agent 运行时的依赖配置."""
    api_key: str
    units: str = "celsius"  # celsius 或 fahrenheit

# ========== 创建 Agent ==========

weather_agent = Agent(
    model="openai:gpt-4o-mini",  # 格式: provider:model
    result_type=WeatherResult,   # 强制输出为 WeatherResult 结构
    deps_type=WeatherDeps,       # 依赖类型
    system_prompt="你是一个精准的天气助手。查询天气后，给出穿衣建议。"
)

# ========== 定义工具（带类型注解）==========

@weather_agent.tool
async def fetch_weather(ctx: RunContext[WeatherDeps], city: str) -> dict:
    """获取指定城市的天气数据.

    Args:
        ctx: 运行上下文，包含依赖配置
        city: 城市名称
    """
    # 模拟天气 API 调用（实际中接入真实 API）
    mock_data = {
        "北京": {"temp": 22, "condition": "晴"},
        "上海": {"temp": 26, "condition": "多云"},
        "深圳": {"temp": 30, "condition": "小雨"},
    }
    data = mock_data.get(city, {"temp": 20, "condition": "未知"})
    return {
        "city": city,
        "temperature": data["temp"],
        "condition": data["condition"],
        "units": ctx.deps.units,  # 从依赖中获取配置
    }

# ========== 运行 Agent ==========

async def main():
    deps = WeatherDeps(api_key="demo-key", units="celsius")

    result = await weather_agent.run(
        "北京今天天气怎么样？",
        deps=deps
    )

    # result.data 已经是 WeatherResult 类型（Pydantic 自动校验）
    print(f"城市: {result.data.city}")
    print(f"温度: {result.data.temperature}°C")
    print(f"天气: {result.data.condition}")
    print(f"建议: {result.data.suggestion}")

# 运行
# import asyncio
# asyncio.run(main())

PydanticAI vs LangChain

维度	PydanticAI	LangChain
核心哲学	类型安全优先	灵活组合优先
学习曲线	低（如果你熟悉 Pydantic）	中（概念较多）
适用场景	生产级、强类型约束项目	快速原型、复杂流程编排
生态成熟度	新兴（2024年底发布）	成熟（社区大、文档多）
推荐人群	TypeScript/Pydantic 爱好者	需要快速上手的开发者

评估最佳实践

1. 准备测试集：收集 50-200 个真实用户问题作为评估基准
2. 人工标注：初期人工评估 10-20 个回答，形成质量认知
3. 自动化回归：每次修改 Prompt 或工具后，自动跑评估集
4. 关注"坏"案例：失败案例比成功案例更有价值
5. 使用 LangSmith / Langfuse：专业的 LLM 可观测性平台，记录和评估每次调用