第7章：智能体（Agents）

在前面的章节中，我们学习了如何使用Chain构建固定的处理流程。但在很多实际场景中，AI需要自主决策——根据用户的问题，智能地选择使用哪些工具、以什么顺序执行。这就是Agent（智能体）的核心价值所在。

1. 什么是Agent

Agent（智能体）是一种能够自主决策、调用工具的AI系统。与传统Chain的固定流程不同，Agent具有以下特点：

• 自主决策：根据输入和中间结果，动态决定下一步操作
• 工具调用：可以调用外部工具（搜索、计算、数据库等）来获取信息
• 循环执行：通过"思考-行动-观察"循环，直到完成任务
• 灵活应变：能处理复杂、多变的问题场景

2. Agent核心组件

一个完整的Agent系统由三个核心组件构成：

2.1 Tools（工具集）

Tools是Agent可以调用的外部功能。每个工具都有：

• 名称：工具的标识符
• 描述：说明工具的用途（LLM根据描述选择工具）
• 函数：实际执行的代码
• 参数模式：定义工具需要的输入参数

2.2 LLM（决策大脑）

LLM是Agent的"大脑"，负责：

• 理解用户输入和目标
• 根据当前状态决定下一步行动
• 选择合适的工具
• 生成最终回答

2.3 AgentExecutor（执行器）

AgentExecutor是Agent的运行环境，负责：

• 管理"思考-行动-观察"的执行循环
• 调用LLM获取决策
• 执行选中的工具
• 处理工具返回的结果
• 控制执行流程（最大迭代次数、错误处理等）

3. 创建工具（详细）

工具是Agent能力的延伸。LangChain提供了多种创建工具的方式：

3.1 使用@tool装饰器（推荐）

最简单的方式是使用@tool装饰器，它自动从函数名、docstring中提取工具信息：

from langchain.tools import tool
from typing import List, Dict
import random

# 方式1：基本用法 - 自动从docstring提取描述
@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气信息。
    
    Args:
        city: 城市名称，如"北京"、"上海"
        
    Returns:
        该城市的天气状况描述
    """
    # 模拟天气查询（实际应用中调用真实API）
    weathers = ["晴朗", "多云", "小雨", "大雨", "雷阵雨"]
    temps = range(15, 35)
    
    weather = random.choice(weathers)
    temp = random.choice(list(temps))
    
    return f"{city}今天天气{weather}，气温{temp}°C"


# 方式2：自定义工具名称和描述
@tool("calculator", description="执行数学计算，支持加减乘除和复杂表达式")
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    try:
        # 安全计算：限制可用函数
        allowed_names = {
            "abs": abs,
            "max": max,
            "min": min,
            "pow": pow,
            "round": round,
        }
        
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"


# 测试工具
if __name__ == "__main__":
    print(search_weather.run("北京"))
    print(calculate.run("2 + 3 * 4"))
    
    # 查看工具信息
    print(f"\n工具名称: {search_weather.name}")
    print(f"工具描述: {search_weather.description}")
    print(f"参数模式: {search_weather.args}")

3.2 使用StructuredTool创建结构化工具

当工具需要多个复杂参数时，使用StructuredTool可以提供更好的参数验证：

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import Optional
import datetime

# 定义参数模式
class StockQueryInput(BaseModel):
    """股票查询参数"""
    stock_code: str = Field(description="股票代码，如'000001'、'AAPL'")
    start_date: Optional[str] = Field(
        default=None, 
        description="开始日期，格式'YYYY-MM-DD'"
    )
    end_date: Optional[str] = Field(
        default=None,
        description="结束日期，格式'YYYY-MM-DD'"
    )


def query_stock_data(stock_code: str, start_date: str = None, end_date: str = None) -> str:
    """查询股票历史数据"""
    # 模拟股票数据查询
    if not end_date:
        end_date = datetime.datetime.now().strftime("%Y-%m-%d")
    if not start_date:
        start_date = (datetime.datetime.now() - datetime.timedelta(days=30)).strftime("%Y-%m-%d")
    
    # 模拟返回数据
    return f"""
股票代码: {stock_code}
查询区间: {start_date} 至 {end_date}
开盘价: 15.20
收盘价: 16.50
最高价: 16.80
最低价: 15.10
成交量: 1250000股
涨跌幅: +8.55%
"""

# 创建结构化工具
stock_tool = StructuredTool.from_function(
    func=query_stock_data,
    name="stock_query",
    description="查询股票的历史交易数据和价格信息",
    args_schema=StockQueryInput,
    return_direct=False,  # 结果返回给LLM继续处理
)

# 使用示例
result = stock_tool.run({
    "stock_code": "000001",
    "start_date": "2024-01-01"
})
print(result)

3.3 创建数据库查询工具

数据库查询是Agent最常见的应用场景之一：

from langchain.tools import tool
import sqlite3
from typing import Optional

# 创建示例数据库
def init_database():
    """初始化示例数据库"""
    conn = sqlite3.connect(':memory:')  # 内存数据库
    cursor = conn.cursor()
    
    # 创建员工表
    cursor.execute('''
        CREATE TABLE employees (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            department TEXT,
            salary INTEGER,
            hire_date TEXT
        )
    ''')
    
    # 插入示例数据
    employees = [
        (1, '张三', '技术部', 15000, '2020-03-15'),
        (2, '李四', '技术部', 18000, '2019-06-20'),
        (3, '王五', '销售部', 12000, '2021-01-10'),
        (4, '赵六', '销售部', 14000, '2020-09-05'),
        (5, '钱七', '财务部', 16000, '2018-12-01'),
    ]
    cursor.executemany('INSERT INTO employees VALUES (?,?,?,?,?)', employees)
    conn.commit()
    
    return conn

# 全局数据库连接
_db_conn = None

def get_db():
    """获取数据库连接（单例模式）"""
    global _db_conn
    if _db_conn is None:
        _db_conn = init_database()
    return _db_conn


@tool
def query_database(sql: str) -> str:
    """执行SQL查询语句，查询员工数据库。
    
    可用表：
    - employees: 员工表，包含字段 id, name, department, salary, hire_date
    
    Args:
        sql: SQL查询语句，如"SELECT * FROM employees WHERE department='技术部'"
        
    Returns:
        查询结果的JSON格式字符串
    """
    import json
    
    # 安全检查：只允许SELECT语句
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith('SELECT'):
        return "错误：只允许执行SELECT查询语句"
    
    try:
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute(sql)
        
        # 获取列名
        columns = [description[0] for description in cursor.description]
        
        # 获取数据
        rows = cursor.fetchall()
        
        # 转换为字典列表
        results = []
        for row in rows:
            results.append(dict(zip(columns, row)))
        
        return json.dumps(results, ensure_ascii=False, indent=2)
        
    except Exception as e:
        return f"查询错误：{str(e)}"


@tool
def get_table_schema(table_name: str) -> str:
    """获取指定表的结构信息。
    
    Args:
        table_name: 表名称
        
    Returns:
        表的字段结构描述
    """
    try:
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute(f"PRAGMA table_info({table_name})")
        columns = cursor.fetchall()
        
        schema_info = []
        for col in columns:
            schema_info.append(f"- {col[1]} ({col[2]})")
        
        return f"表 {table_name} 的结构：\n" + "\n".join(schema_info)
        
    except Exception as e:
        return f"获取表结构错误：{str(e)}"


# 测试
if __name__ == "__main__":
    print("=== 表结构 ===")
    print(get_table_schema.run("employees"))
    
    print("\n=== 查询技术部员工 ===")
    print(query_database.run("SELECT name, salary FROM employees WHERE department='技术部'"))

3.4 工具描述的重要性

工具描述是Agent选择工具的唯一依据！一个好的工具描述应该包含：

from langchain.tools import tool

# ❌ 不好的示例：描述太简单
@tool
def bad_search(query: str) -> str:
    """搜索功能"""
    pass

# ✅ 好的示例：详细说明用途、参数、使用场景
@tool
def good_search(query: str) -> str:
    """使用Google搜索引擎查询最新信息。
    
    适用于：
    - 查询当前时间、天气、新闻等实时信息
    - 获取最新的事件、股价、比赛结果
    - 查找特定的网页内容
    
    Args:
        query: 搜索关键词，建议具体明确
        
    Returns:
        搜索结果的摘要信息，包含前3条结果
        
    示例：
        "2024年美国总统大选结果"
        "今日上证指数收盘"
    """
    pass

4. Agent类型

LangChain提供了多种Agent类型，适用于不同场景：

4.1 zero-shot-react-description（最常用）

根据工具描述直接选择工具，适合大多数场景：

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain import hub
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# 定义工具
@tool
def search(query: str) -> str:
    """搜索最新信息。用于查询实时数据、新闻、天气等。"""
    # 模拟搜索
    return f"搜索结果：关于'{query}'的最新信息显示..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。支持加减乘除和括号。"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except:
        return "计算失败，请检查表达式"

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

# 工具列表
tools = [
    Tool(
        name="search",
        func=search.run,
        description="搜索最新信息。用于查询实时数据、新闻、天气等。"
    ),
    Tool(
        name="calculator",
        func=calculator.run,
        description="计算数学表达式。支持加减乘除和括号。"
    ),
    Tool(
        name="get_time",
        func=get_current_time.run,
        description="获取当前日期和时间。"
    ),
]

# 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 获取ReAct提示词模板
prompt = hub.pull("hwchase17/react")

# 创建Agent
agent = create_react_agent(llm, tools, prompt)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印执行过程
    max_iterations=10,  # 最大迭代次数
    handle_parsing_errors=True,  # 处理解析错误
)

# 运行Agent
response = agent_executor.invoke({
    "input": "现在几点了？北京今天的天气如何？"
})
print(response["output"])

4.2 structured-chat（结构化输入）

支持结构化输入的Agent，适合需要复杂参数的工具：

from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain import hub
from langchain_openai import ChatOpenAI
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

# 定义复杂参数的工具
class SendEmailInput(BaseModel):
    to: str = Field(description="收件人邮箱地址")
    subject: str = Field(description="邮件主题")
    content: str = Field(description="邮件正文内容")
    cc: list = Field(default=[], description="抄送列表")

def send_email(to: str, subject: str, content: str, cc: list = None) -> str:
    """发送邮件"""
    # 模拟发送
    return f"邮件已发送至 {to}，主题：{subject}"

email_tool = StructuredTool.from_function(
    func=send_email,
    name="send_email",
    description="发送邮件给指定收件人",
    args_schema=SendEmailInput,
)

# 创建结构化Agent
tools = [email_tool]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/structured-chat-agent")

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

# 使用
response = agent_executor.invoke({
    "input": "给张三发一封邮件，告诉他会议改到明天下午3点"
})

4.3 conversational（带对话记忆的Agent）

保留对话历史的Agent，适合多轮对话场景：

from langchain.agents import create_conversational_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# 定义工具
@tool
def get_user_info(user_id: str) -> str:
    """获取用户信息"""
    users = {
        "u001": {"name": "张三", "level": "VIP", "points": 1250},
        "u002": {"name": "李四", "level": "普通", "points": 350},
    }
    user = users.get(user_id, {})
    return str(user)

@tool
def query_orders(user_id: str) -> str:
    """查询用户订单"""
    return f"用户 {user_id} 最近订单：订单A、订单B"

# 创建记忆组件
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建对话Agent
tools = [
    Tool(name="get_user_info", func=get_user_info.run, description="获取用户信息"),
    Tool(name="query_orders", func=query_orders.run, description="查询用户订单"),
]

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
agent = create_conversational_agent(llm, tools)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 多轮对话示例
print("=== 第一轮 ===")
response1 = agent_executor.invoke({"input": "帮我查一下用户u001的信息"})
print(response1["output"])

print("\n=== 第二轮（利用记忆）===")
response2 = agent_executor.invoke({"input": "那他最近有什么订单？"})
print(response2["output"])

5. AgentExecutor详解

AgentExecutor是Agent的运行容器，负责管理执行流程。

5.1 执行循环

AgentExecutor的执行流程如下：

# Agent执行流程示意
"""
循环开始
  ↓
LLM根据输入+历史决定：
  - 使用哪个Action（工具）
  - 或给出Final Answer（结束）
  ↓
如果是Action：
  - 执行工具
  - 获取Observation（结果）
  - 记录到历史
  - 回到循环开始
  ↓
如果是Final Answer：
  - 返回答案
  - 结束循环
"""

5.2 配置参数

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,              # Agent实例
    tools=tools,              # 工具列表
    
    # 执行控制
    max_iterations=15,        # 最大迭代次数，防止无限循环
    max_execution_time=60,    # 最大执行时间（秒）
    
    # 错误处理
    handle_parsing_errors=True,  # 自动处理LLM输出解析错误
    
    # 输出选项
    verbose=True,             # 打印详细执行过程
    return_intermediate_steps=True,  # 返回中间步骤
    
    # 提前停止
    early_stopping_method="generate",  # 达到限制时如何处理
)

# 获取详细执行信息
response = agent_executor.invoke(
    {"input": "查询问题"},
    return_only_outputs=False
)

# 如果设置了return_intermediate_steps=True
print("最终答案:", response["output"])
print("执行步骤:", response["intermediate_steps"])

5.3 处理工具错误

工具执行失败时，Agent应该能够优雅处理：

from langchain.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from langchain_openai import ChatOpenAI

# 带有错误处理的工具
@tool
def safe_api_call(endpoint: str) -> str:
    """调用API获取数据"""
    try:
        # 模拟API调用
        if "error" in endpoint:
            raise Exception("API返回错误：连接超时")
        return f"API {endpoint} 返回数据: {{'status': 'ok'}}"
    except Exception as e:
        # 返回错误信息给Agent，让它决定如何处理
        return f"工具执行错误：{str(e)}。请尝试其他方式或告知用户。"

@tool
def fallback_search(query: str) -> str:
    """备用搜索"""
    return f"通过备用搜索找到关于'{query}'的信息"

# 创建Agent
tools = [safe_api_call, fallback_search]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
)

# 测试：当API失败时，Agent会尝试备用方案
result = agent_executor.invoke({
    "input": "调用error端点获取数据"
})

6. 自定义Agent

LangChain允许深度定制Agent的各个组件。

6.1 自定义提示词模板

通过自定义提示词，可以引导Agent按特定方式工作：

from langchain.agents import create_react_agent, AgentExecutor
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# 自定义ReAct提示词
CUSTOM_REACT_TEMPLATE = """你是一个专业的数据分析助手，按照以下规则工作：

## 可用工具
{tools}

## 工具名称
{tool_names}

## 工作原则
1. 首先分析用户问题的类型
2. 如果需要计算，优先使用calculator工具
3. 如果需要实时信息，使用search工具
4. 每一步都要解释你的思考过程

## 输出格式
问题: 需要回答的问题
思考: 分析问题并决定下一步行动
行动: 要使用的工具名称，必须是[{tool_names}]之一
行动输入: 传递给工具的输入
观察: 工具执行的结果
...(重复思考/行动/行动输入/观察)

思考: 我现在知道最终答案
最终答案: 对原始问题的回答

开始！

问题: {input}
思考: {agent_scratchpad}"""

# 创建自定义提示词
prompt = PromptTemplate.from_template(CUSTOM_REACT_TEMPLATE).partial(
    tools="",  # 将在运行时填充
    tool_names="",  # 将在运行时填充
)

# 定义工具
@tool
def calculator(expression: str) -> str:
    """数学计算"""
    return str(eval(expression))

@tool
def search(query: str) -> str:
    """信息搜索"""
    return f"搜索结果：{query}"

tools = [calculator, search]

# 创建Agent（需要手动填充tools和tool_names）
llm = ChatOpenAI(model="gpt-4o", temperature=0)

from langchain.agents.format_scratchpad import format_log_to_str
from langchain.agents.output_parsers import ReActSingleInputOutputParser

# 构建完整提示词
tool_strings = "\n".join([f"{tool.name}: {tool.description}" for tool in tools])
tool_names = ", ".join([tool.name for tool in tools])

prompt = PromptTemplate.from_template(CUSTOM_REACT_TEMPLATE).partial(
    tools=tool_strings,
    tool_names=tool_names,
)

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

# 使用
result = agent_executor.invoke({"input": "计算 123 * 456 并搜索Python最新版本"})

6.2 自定义解析逻辑

处理LLM输出的特殊情况：

from langchain.agents import AgentOutputParser
from langchain.schema import AgentAction, AgentFinish
from typing import Union
import re

class CustomOutputParser(AgentOutputParser):
    """自定义输出解析器，处理特殊格式"""
    
    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        # 检查是否包含最终答案
        if "最终答案：" in text or "Final Answer:" in text:
            # 提取最终答案
            patterns = [
                r"最终答案：(.*)$",
                r"Final Answer: (.*)$"
            ]
            for pattern in patterns:
                match = re.search(pattern, text, re.DOTALL)
                if match:
                    return AgentFinish(
                        return_values={"output": match.group(1).strip()},
                        log=text
                    )
        
        # 解析行动
        action_patterns = [
            r"行动[：:]\s*(\w+)\s*\n行动输入[：:]\s*(.*?)(?=\n观察|$)",
            r"Action:\s*(\w+)\s*\nAction Input:\s*(.*?)(?=\nObservation|$)"
        ]
        
        for pattern in action_patterns:
            match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
            if match:
                action = match.group(1).strip()
                action_input = match.group(2).strip()
                return AgentAction(
                    tool=action,
                    tool_input=action_input,
                    log=text
                )
        
        # 无法解析时返回特定格式
        return AgentFinish(
            return_values={"output": f"无法解析的响应：{text}"},
            log=text
        )
    
    @property
    def _type(self) -> str:
        return "custom"

# 使用自定义解析器
# 注意：这需要配合自定义Agent使用，这里仅展示解析器实现

7. 完整实战：数据分析助手Agent

让我们构建一个完整的数据分析助手Agent，它能够：

• 搜索最新数据
• 进行数学计算和统计分析
• 生成可视化图表

"""
数据分析助手Agent
功能：搜索数据、计算统计量、生成图表
"""

import os
import json
from typing import List, Dict
from datetime import datetime

from langchain.agents import create_react_agent, AgentExecutor, Tool
from langchain import hub
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# 设置环境变量
# os.environ["OPENAI_API_KEY"] = "your-api-key"

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

@tool
def search_economic_data(indicator: str, year: str = "2023") -> str:
    """搜索经济指标数据。
    
    Args:
        indicator: 指标名称，如"GDP"、"CPI"、"失业率"
        year: 年份，如"2023"、"2022-2023"
        
    Returns:
        该指标的年度数据（JSON格式）
    """
    # 模拟经济数据（实际应用应调用真实API）
    mock_data = {
        "GDP": {
            "2020": {"value": 101.6, "unit": "万亿元", "growth": 2.3},
            "2021": {"value": 114.9, "unit": "万亿元", "growth": 8.4},
            "2022": {"value": 121.0, "unit": "万亿元", "growth": 3.0},
            "2023": {"value": 126.0, "unit": "万亿元", "growth": 5.2},
        },
        "CPI": {
            "2020": {"value": 102.5, "unit": "上年=100", "growth": 2.5},
            "2021": {"value": 100.9, "unit": "上年=100", "growth": 0.9},
            "2022": {"value": 102.0, "unit": "上年=100", "growth": 2.0},
            "2023": {"value": 100.2, "unit": "上年=100", "growth": 0.2},
        }
    }
    
    if indicator in mock_data:
        if "-" in year:
            # 返回区间数据
            start, end = year.split("-")
            years = range(int(start), int(end) + 1)
            result = {str(y): mock_data[indicator].get(str(y), {}) for y in years}
        else:
            result = {year: mock_data[indicator].get(year, {})}
        return json.dumps(result, ensure_ascii=False)
    
    return f"未找到 {indicator} 的数据"


@tool
def calculate_statistics(data_json: str, operation: str = "growth_rate") -> str:
    """对数据进行统计分析。
    
    Args:
        data_json: JSON格式的数据数组或字典
        operation: 统计操作，可选：
            - growth_rate: 计算增长率
            - average: 计算平均值
            - sum: 计算总和
            - max: 最大值
            - min: 最小值
            
    Returns:
        统计结果
    """
    try:
        data = json.loads(data_json)
        
        # 提取数值
        if isinstance(data, dict):
            values = [v.get("value", v) for v in data.values() if isinstance(v, dict)]
            if not values:
                values = list(data.values())
        else:
            values = data
        
        # 确保是数字列表
        values = [float(v) for v in values if isinstance(v, (int, float, str))]
        
        if operation == "growth_rate" and len(values) >= 2:
            # 计算复合增长率
            total_growth = (values[-1] - values[0]) / values[0] * 100
            years = len(values) - 1
            cagr = ((values[-1] / values[0]) ** (1/years) - 1) * 100 if years > 0 else 0
            return f"总增长率: {total_growth:.2f}%, 年均复合增长率(CAGR): {cagr:.2f}%"
        
        elif operation == "average":
            return f"平均值: {sum(values)/len(values):.2f}"
        elif operation == "sum":
            return f"总和: {sum(values):.2f}"
        elif operation == "max":
            return f"最大值: {max(values):.2f}"
        elif operation == "min":
            return f"最小值: {min(values):.2f}"
        else:
            return f"不支持的统计操作: {operation}"
            
    except Exception as e:
        return f"计算错误: {str(e)}"


@tool
def generate_chart(data_json: str, chart_type: str = "line", title: str = "数据图表") -> str:
    """生成数据可视化图表。
    
    Args:
        data_json: JSON格式数据，如 {"2020": 101.6, "2021": 114.9}
        chart_type: 图表类型，可选 line(折线图)、bar(柱状图)
        title: 图表标题
        
    Returns:
        图表保存路径或描述
    """
    try:
        import matplotlib.pyplot as plt
        import matplotlib
        matplotlib.use('Agg')  # 非交互式后端
        
        data = json.loads(data_json)
        
        # 准备数据
        labels = list(data.keys())
        if isinstance(list(data.values())[0], dict):
            values = [v.get("value", 0) for v in data.values()]
        else:
            values = [float(v) for v in data.values()]
        
        # 创建图表
        fig, ax = plt.subplots(figsize=(10, 6))
        
        if chart_type == "line":
            ax.plot(labels, values, marker='o', linewidth=2, markersize=8)
        elif chart_type == "bar":
            ax.bar(labels, values, color='steelblue')
        
        ax.set_title(title, fontsize=16, fontweight='bold')
        ax.set_xlabel('年份', fontsize=12)
        ax.set_ylabel('数值', fontsize=12)
        ax.grid(True, alpha=0.3)
        
        # 添加数值标签
        for i, (label, value) in enumerate(zip(labels, values)):
            ax.annotate(f'{value}', xy=(i, value), ha='center', va='bottom')
        
        plt.tight_layout()
        
        # 保存图表
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"chart_{timestamp}.png"
        plt.savefig(filename, dpi=150, bbox_inches='tight')
        plt.close()
        
        return f"图表已生成并保存至: {filename}"
        
    except Exception as e:
        return f"生成图表失败: {str(e)}。请确保已安装matplotlib: pip install matplotlib"


@tool
def format_report(title: str, sections: str) -> str:
    """格式化生成分析报告。
    
    Args:
        title: 报告标题
        sections: JSON格式的报告内容，如：
            {"背景": "...", "数据分析": "...", "结论": "..."}
        
    Returns:
        格式化后的报告文本
    """
    try:
        sections_dict = json.loads(sections)
        
        report = f"""
{'=' * 60}
{title.center(60)}
{'=' * 60}
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

"""
        for section_name, content in sections_dict.items():
            report += f"【{section_name}】\n{content}\n\n"
        
        report += "=" * 60 + "\n"
        
        return report
        
    except Exception as e:
        return f"格式化报告失败: {str(e)}"


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

def create_data_analysis_agent():
    """创建数据分析助手Agent"""
    
    # 工具列表
    tools = [
        Tool(
            name="search_economic_data",
            func=search_economic_data.run,
            description="""搜索经济指标数据，如GDP、CPI等。
参数：indicator(指标名), year(年份或年份区间如"2020-2023")
返回：JSON格式的年度数据"""
        ),
        Tool(
            name="calculate_statistics",
            func=calculate_statistics.run,
            description="""对数据进行统计分析。
参数：data_json(数据), operation(操作：growth_rate/average/sum/max/min)
返回：统计结果"""
        ),
        Tool(
            name="generate_chart",
            func=generate_chart.run,
            description="""生成数据可视化图表。
参数：data_json(数据), chart_type(图表类型:line/bar), title(标题)
返回：图表保存路径"""
        ),
        Tool(
            name="format_report",
            func=format_report.run,
            description="""格式化生成分析报告。
参数：title(标题), sections(JSON格式的内容章节)
返回：格式化报告"""
        ),
    ]
    
    # 初始化LLM
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0,  # 分析任务需要确定性
    )
    
    # 获取ReAct提示词
    prompt = hub.pull("hwchase17/react")
    
    # 创建Agent
    agent = create_react_agent(llm, tools, prompt)
    
    # 创建执行器
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=10,
        handle_parsing_errors=True,
        return_intermediate_steps=True,
    )
    
    return agent_executor


# ==================== 主程序 ====================

if __name__ == "__main__":
    # 创建Agent
    agent = create_data_analysis_agent()
    
    # 测试查询
    queries = [
        "查询2020-2023年的GDP数据，计算增长率，并生成趋势图",
        "获取2023年CPI数据，并生成分析报告",
    ]
    
    for query in queries:
        print(f"\n{'='*60}")
        print(f"用户问题: {query}")
        print('='*60)
        
        try:
            result = agent.invoke({"input": query})
            print(f"\n📊 最终答案:\n{result['output']}")
            
            # 显示执行步骤
            if 'intermediate_steps' in result:
                print(f"\n📝 执行了 {len(result['intermediate_steps'])} 个步骤")
                
        except Exception as e:
            print(f"❌ 执行出错: {e}")

8. Agent与Chain的区别

理解什么时候用Agent，什么时候用Chain，是设计AI应用的关键。

8.1 核心区别对比

8.2 选择建议

使用Chain的场景：

• ✅ 流程固定且明确（如：翻译→润色→输出）
• ✅ 对成本和延迟敏感
• ✅ 需要高度可控的输出
• ✅ 每个步骤的输出格式固定

使用Agent的场景：

• ✅ 问题开放，解决路径不明确
• ✅ 需要根据中间结果动态决策
• ✅ 可能需要调用多个外部工具
• ✅ 问题类型多变，难以预先定义流程

8.3 混合使用

最佳实践是Chain和Agent结合使用：

from langchain.chains import LLMChain
from langchain.agents import create_react_agent, AgentExecutor
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 场景：客服系统
# - 先用Chain进行意图分类（固定流程）
# - 再根据意图选择处理方式

# 1. 意图分类Chain
classify_prompt = PromptTemplate.from_template("""
分析用户问题，判断意图类型：
- question: 一般性问题
- order_query: 订单查询
- complaint: 投诉
- refund: 退款申请

用户问题: {question}
意图类型:
""")

llm = ChatOpenAI(model="gpt-3.5-turbo")
classify_chain = LLMChain(llm=llm, prompt=classify_prompt)

# 2. 订单查询Agent（复杂场景）
order_tools = [...]  # 订单相关工具
order_agent = create_react_agent(llm, order_tools, ...)

# 3. 主流程
question = "我的订单为什么还没发货？"
intent = classify_chain.run(question).strip()

if intent == "order_query":
    # 使用Agent处理复杂查询
    response = order_agent.invoke({"input": question})
elif intent == "question":
    # 使用简单Chain回答
    response = simple_qa_chain.run(question)
else:
    # 转人工
    response = "正在为您转接人工客服..."

本章小结

在本章中，我们深入学习了LangChain的Agent系统：

1. Agent概念：自主决策、调用工具的AI系统
2. 核心组件：Tools、LLM、AgentExecutor
3. 创建工具：@tool装饰器、StructuredTool、数据库工具
4. Agent类型：zero-shot-react、structured-chat、conversational
5. AgentExecutor：执行循环、参数配置、错误处理
6. 自定义Agent：自定义提示词、自定义解析逻辑
7. 实战项目：数据分析助手Agent
8. Chain vs Agent：选择策略和混合使用