Jersey's Blog

学习时长：2-3 周

Agent（智能体）是 AI 应用开发中最具潜力的方向之一,它让 LLM 从"被动回答"进化为"主动解决问题"。本节将深入讲解 Agent 的核心原理、架构设计和实现方法。

5.1.1 什么是 Agent?

定义与特征

Agent 的本质

Agent 是一个能够感知环境、自主决策、执行行动并达成目标的智能系统。与传统的单次问答不同,Agent 具备以下核心特征:

特征	传统 LLM 对话	Agent 系统
交互模式	单轮问答	多轮迭代,持续执行
工具使用	❌ 无法调用外部工具	✅ 主动调用 API、数据库、搜索引擎
规划能力	❌ 只回答当前问题	✅ 分解任务,制定执行计划
记忆机制	❌ 仅限对话上下文	✅ 长期记忆、知识库检索
自主性	❌ 完全依赖用户指令	✅ 自主决策下一步行动
反思能力	❌ 不会评估输出质量	✅ 自我反思、错误修正

Agent 的核心能力

用户目标
    ↓
【感知】理解任务需求
    ↓
【规划】分解为子任务 → 制定执行计划
    ↓
【执行】循环执行:
    ├─ 决策: 选择下一步行动(调用工具/生成回复)
    ├─ 行动: 执行工具调用或生成内容
    ├─ 观察: 获取执行结果
    └─ 反思: 评估是否达成目标
    ↓
【输出】最终结果

Agent vs 传统编程

示例对比: 查询天气并推荐穿衣

传统编程方式:

python

# 固定流程,无法应对变化
def weather_assistant(city: str):
    # 1. 调用天气 API
    weather = call_weather_api(city)
    
    # 2. 硬编码规则
    if weather['temp'] < 10:
        advice = "建议穿羽绒服"
    elif weather['temp'] < 20:
        advice = "建议穿外套"
    else:
        advice = "建议穿短袖"
    
    return f"{city}今天{weather['temp']}°C,{advice}"

Agent 方式:

python

# Agent 自主决策,灵活应对
agent = Agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[
        WeatherTool(),      # 查询天气
        CalendarTool(),     # 查询日程
        ClothingDBTool()    # 查询衣物数据库
    ]
)

# Agent 自主规划:
# 1. 调用 WeatherTool 查询天气
# 2. 发现用户有户外活动,调用 CalendarTool
# 3. 综合考虑天气+活动类型,调用 ClothingDBTool
# 4. 生成个性化建议: "今天15°C有小雨,下午有户外会议,
#    建议穿防水外套+商务休闲装"

result = agent.run("帮我看看今天穿什么合适")

关键差异:

✅ 灵活性: Agent 可根据实际情况调整执行路径
✅ 可扩展: 新增工具无需修改核心逻辑
✅ 智能化: 理解隐含需求(如考虑日程安排)

5.1.2 Agent 核心架构

ReAct 模式 (Reasoning + Acting)

ReAct 是目前最主流的 Agent 架构,由 Google 在 2022 年提出。核心思想是推理与行动交替进行。

ReAct 循环流程:

用户输入: "帮我订一张明天北京到上海的机票"

【Thought 1】我需要先查询明天的航班信息
【Action 1】Tool: search_flights, Input: {"from": "北京", "to": "上海", "date": "2024-03-15"}
【Observation 1】找到 3 个航班: CA1234(08:00, ¥800), MU5678(14:00, ¥650), ...

【Thought 2】用户没有指定时间和预算,我应该询问偏好
【Action 2】Tool: ask_user, Input: {"question": "您偏好的出发时间和预算是?"}
【Observation 2】用户回复: "下午出发,预算 700 元以内"

【Thought 3】根据用户需求,MU5678 最合适,现在执行订票
【Action 3】Tool: book_flight, Input: {"flight_id": "MU5678", "passengers": 1}
【Observation 3】订票成功,订单号: ORDER123456

【Thought 4】任务完成,向用户确认
【Final Answer】已为您预订明天 14:00 的 MU5678 航班(¥650),订单号 ORDER123456

ReAct Prompt 模板:

python

REACT_PROMPT = """
你是一个智能助手,可以使用以下工具来完成任务:

{tools_description}

请使用以下格式回答:

Thought: 分析当前情况,思考下一步应该做什么
Action: 要执行的工具名称
Action Input: 工具的输入参数(JSON 格式)
Observation: 工具执行的结果
... (重复 Thought/Action/Observation 直到找到答案)
Thought: 我现在知道最终答案了
Final Answer: 最终答案

开始!

用户问题: {input}

{agent_scratchpad}
"""

核心组件详解

1. LLM (大语言模型) - 大脑

Agent 的决策中枢,负责:

理解用户意图
分解任务
选择工具
生成最终回复

模型选择建议:

任务复杂度	推荐模型	原因
简单(单工具调用)	GPT-4o-mini, Qwen-Plus	成本低,速度快
中等(多步推理)	GPT-4o, Claude-3.5-Sonnet	推理能力强
复杂(长链规划)	GPT-4, Claude-3-Opus	顶级推理,少出错
代码生成	Claude-3.5-Sonnet, DeepSeek-Coder	代码能力强

2. Tools (工具集) - 手脚

赋予 Agent 与外部世界交互的能力。

工具定义标准:

python

from pydantic import BaseModel, Field

class ToolInput(BaseModel):
    """工具输入参数的 Schema"""
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="返回结果数量")

class Tool:
    name: str = "search_web"
    description: str = "在互联网上搜索信息,适合查询实时资讯、百科知识"
    args_schema: type[BaseModel] = ToolInput
    
    def run(self, query: str, max_results: int = 5) -> str:
        """执行工具逻辑"""
        results = search_api(query, limit=max_results)
        return format_results(results)

常见工具类型:

类型	示例	用途
搜索工具	Google Search, Bing API	获取实时信息
计算工具	Python REPL, WolframAlpha	数学计算、代码执行
数据库工具	SQL Query, Vector Search	查询结构化/非结构化数据
API 工具	天气 API, 股票 API	调用第三方服务
文件工具	Read/Write File	文件读写操作
通信工具	Send Email, Slack Bot	发送消息通知

3. Memory (记忆系统) - 经验

Agent 需要记住历史交互和学到的知识。

记忆类型:

python

class AgentMemory:
    # 短期记忆: 当前对话的上下文
    short_term: list[Message] = []
    
    # 长期记忆: 持久化的知识和经验
    long_term: VectorStore = ChromaDB()
    
    # 工作记忆: 当前任务的中间状态
    working_memory: dict = {
        "current_goal": "订机票",
        "subtasks": ["查询航班", "确认偏好", "执行订票"],
        "completed": ["查询航班"]
    }

记忆检索策略:

python

def retrieve_relevant_memory(query: str, k: int = 3) -> list[str]:
    """从长期记忆中检索相关经验"""
    # 1. 向量化查询
    query_embedding = get_embedding(query)
    
    # 2. 相似度检索
    results = memory_db.similarity_search(
        query_embedding, 
        k=k,
        filter={"type": "successful_execution"}  # 只检索成功案例
    )
    
    # 3. 格式化为 Prompt
    return [f"历史经验 {i+1}: {r.content}" for i, r in enumerate(results)]

4. Planning (规划模块) - 策略

将复杂任务分解为可执行的子任务。

规划方法对比:

方法	原理	优点	缺点
ReAct	边思考边行动	灵活,适应性强	可能绕弯路
Plan-and-Execute	先规划后执行	结构清晰,高效	难以应对变化
Tree of Thoughts	树状搜索多路径	探索充分,质量高	计算成本高
Reflexion	执行后反思改进	自我修正能力强	需要多轮迭代

5.1.3 Agent 类型与应用场景

按能力分类

1. 单 Agent (Single Agent)

一个 Agent 独立完成任务。

适用场景:

简单的信息查询(天气、新闻)
单一领域的专家系统(法律咨询、医疗问诊)
个人助理(日程管理、邮件处理)

示例: 数据分析 Agent

python

data_analyst = Agent(
    role="数据分析师",
    goal="分析销售数据并生成报告",
    tools=[
        PythonREPLTool(),      # 执行数据分析代码
        PlotlyTool(),          # 生成可视化图表
        ExcelTool()            # 读写 Excel 文件
    ],
    llm=ChatOpenAI(model="gpt-4o")
)

result = data_analyst.run(
    "分析 sales_2024.xlsx,找出销量下降的原因并给出建议"
)

2. 多 Agent 协作 (Multi-Agent System)

多个 Agent 分工协作,完成复杂任务。

协作模式:

【层级协作】
  Manager Agent (协调者)
      ↓
  ├─ Research Agent (研究员)
  ├─ Writer Agent (写作者)
  └─ Reviewer Agent (审核者)

【平等协作】
  Agent A ←→ Agent B ←→ Agent C
  (通过消息传递协作)

示例: 内容创作团队

python

from crewai import Crew, Agent, Task

# 定义 Agent 团队
researcher = Agent(
    role="研究员",
    goal="收集关于 {topic} 的最新资料",
    tools=[SearchTool(), ScraperTool()]
)

writer = Agent(
    role="内容创作者",
    goal="根据研究资料撰写高质量文章",
    tools=[LanguageTool()]
)

editor = Agent(
    role="编辑",
    goal="审核文章质量并提出修改建议",
    tools=[GrammarCheckTool()]
)

# 定义任务流
task1 = Task(description="研究 AI Agent 的最新进展", agent=researcher)
task2 = Task(description="撰写 2000 字科普文章", agent=writer)
task3 = Task(description="审核并优化文章", agent=editor)

# 执行协作
crew = Crew(agents=[researcher, writer, editor], tasks=[task1, task2, task3])
result = crew.kickoff(inputs={"topic": "AI Agent"})

3. 自主 Agent (Autonomous Agent)

长期运行,持续监控环境并主动执行任务。

特点:

🔄 持续运行(守护进程)
📊 主动监控环境变化
⚡ 触发条件满足时自动执行
🎯 自主设定子目标

示例: 股票交易 Agent

python

class TradingAgent:
    def __init__(self):
        self.portfolio = Portfolio()
        self.strategy = MomentumStrategy()
        self.risk_limit = 0.02  # 单次最大亏损 2%
    
    async def run_forever(self):
        """持续运行的主循环"""
        while True:
            # 1. 监控市场
            market_data = await self.fetch_market_data()
            
            # 2. 分析信号
            signals = self.strategy.analyze(market_data)
            
            # 3. 决策
            if signals['buy'] and self.risk_check():
                await self.execute_trade('BUY', signals['symbol'])
            elif signals['sell']:
                await self.execute_trade('SELL', signals['symbol'])
            
            # 4. 记录日志
            self.log_state()
            
            # 5. 等待下一个周期
            await asyncio.sleep(60)  # 每分钟检查一次

按应用领域分类

1. 任务执行型 Agent

特点: 执行具体的操作任务(订票、发邮件、写代码)

案例: GitHub Copilot Workspace

自动分析 Issue
生成代码修复方案
创建 Pull Request
运行测试验证

2. 知识问答型 Agent

特点: 结合 RAG 检索知识库,提供专业解答

案例: 企业智能客服

检索产品文档
查询订单状态
解答技术问题
升级到人工客服(必要时)

3. 创作型 Agent

特点: 生成文本、图片、视频等创意内容

案例: 自动化视频制作

根据主题生成脚本
调用 TTS 生成旁白
调用 Sora 生成视频片段
使用 FFmpeg 合成最终视频

4. 分析决策型 Agent

特点: 分析数据,提供决策建议

案例: 商业智能助手

分析销售数据趋势
预测未来市场需求
推荐营销策略
生成可视化报告

5.1.4 Agent 开发框架对比

主流框架选型

框架	开发者	特点	适用场景	学习曲线
LangGraph	LangChain	图状态机,灵活可控	复杂流程,需要精细控制	中
AutoGPT	Significant Gravitas	全自主,长期运行	探索性任务,研究实验	低
BabyAGI	Yohei Nakajima	轻量级,任务驱动	简单自主任务	低
CrewAI	CrewAI Inc	多 Agent 协作	团队协作场景	中
MetaGPT	DeepWisdom	软件开发流程	代码生成,项目管理	高
Semantic Kernel	Microsoft	企业级,插件系统	.NET 生态,企业应用	中

快速选型指南:

需求: 我想开发一个...

├─ 简单的工具调用(查天气、搜索)
│   → 使用 LangChain Agent 即可
│
├─ 复杂的多步骤流程(需要条件分支、循环)
│   → LangGraph (推荐)
│
├─ 多个 Agent 协作(如内容创作团队)
│   → CrewAI
│
├─ 长期自主运行的任务(如自动化运维)
│   → AutoGPT / BabyAGI
│
└─ 软件开发相关(生成代码、项目管理)
    → MetaGPT

LangGraph 快速入门

LangGraph 是目前最灵活的 Agent 框架,基于状态图(State Graph)设计。

核心概念:

python

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息历史
    next_action: str                         # 下一步行动
    tool_results: dict                       # 工具执行结果

# 2. 定义节点(每个节点是一个函数)
def call_model(state: AgentState) -> AgentState:
    """调用 LLM 决策"""
    response = llm.invoke(state["messages"])
    return {"messages": [response], "next_action": parse_action(response)}

def execute_tool(state: AgentState) -> AgentState:
    """执行工具"""
    tool_name = state["next_action"]
    result = tools[tool_name].run(state)
    return {"tool_results": {tool_name: result&#125;&#125;

def should_continue(state: AgentState) -> str:
    """条件分支:决定继续还是结束"""
    if "Final Answer" in state["messages"][-1].content:
        return "end"
    return "continue"

# 3. 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", execute_tool)

graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "tools", "end": END}
)
graph.add_edge("tools", "agent")

# 4. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [("user", "帮我查询北京今天的天气")]})

LangGraph 可视化:

    [用户输入]
        ↓
    ┌─────────┐
    │  Agent  │ ← 调用 LLM 决策
    └─────────┘
        ↓
    [条件判断]
    ├─ 需要工具 → ┌────────┐
    │              │ Tools  │ ← 执行工具
    │              └────────┘
    │                  ↓
    │              (循环回 Agent)
    │
    └─ 有最终答案 → [END]

5.1.5 Agent 设计最佳实践

1. Prompt 工程

清晰的角色定义:

python

AGENT_SYSTEM_PROMPT = """
你是一个专业的数据分析师 Agent,具备以下能力:

【核心职责】
- 分析结构化数据(CSV、Excel、SQL 数据库)
- 生成数据可视化图表
- 撰写分析报告

【工作流程】
1. 理解用户的分析需求
2. 选择合适的工具获取数据
3. 使用 Python 进行数据处理和分析
4. 生成图表和报告
5. 向用户解释分析结果

【工具使用规范】
- 使用 `python_repl` 执行数据分析代码
- 使用 `sql_query` 查询数据库
- 使用 `plot_chart` 生成可视化图表

【输出要求】
- 分析结论要有数据支撑
- 图表要清晰易懂
- 报告要结构化(使用 Markdown 格式)

【注意事项】
- 如果数据不足,主动询问用户
- 如果发现数据异常,及时提醒
- 不要对数据做主观臆断
"""

结构化输出:

python

from pydantic import BaseModel, Field

class AnalysisReport(BaseModel):
    """强制 Agent 输出结构化报告"""
    summary: str = Field(description="分析摘要(100字以内)")
    key_findings: list[str] = Field(description="关键发现(3-5条)")
    visualizations: list[str] = Field(description="生成的图表路径")
    recommendations: list[str] = Field(description="行动建议")
    confidence_score: float = Field(ge=0, le=1, description="结论置信度")

# 在 Prompt 中要求输出 JSON
prompt = f"""
请按照以下 JSON Schema 输出分析报告:
{AnalysisReport.schema_json(indent=2)}
"""

2. 工具设计原则

单一职责:

python

# ❌ 不好: 工具功能过于复杂
class SuperTool:
    def run(self, action: str, **kwargs):
        if action == "search":
            return self.search(**kwargs)
        elif action == "calculate":
            return self.calculate(**kwargs)
        elif action == "send_email":
            return self.send_email(**kwargs)
        # ... 太多功能

# ✅ 好: 每个工具只做一件事
class SearchTool:
    name = "search_web"
    description = "在互联网上搜索信息"
    def run(self, query: str) -> str:
        return search_api(query)

class CalculatorTool:
    name = "calculator"
    description = "执行数学计算"
    def run(self, expression: str) -> float:
        return eval(expression)

清晰的描述:

python

class WeatherTool:
    name = "get_weather"
    
    # ❌ 模糊的描述
    description = "获取天气"
    
    # ✅ 清晰的描述
    description = """
    查询指定城市的实时天气信息。
    
    返回内容包括:
    - 温度(摄氏度)
    - 天气状况(晴/雨/雪等)
    - 湿度百分比
    - 风力等级
    
    适用场景:
    - 用户询问某地天气
    - 需要天气数据做决策(如穿衣建议、出行规划)
    
    注意: 只能查询中国大陆城市,不支持国外城市
    """

错误处理:

python

class RobustTool:
    def run(self, **kwargs) -> str:
        try:
            result = self._execute(**kwargs)
            return f"执行成功: {result}"
        except ValueError as e:
            return f"参数错误: {e}. 请检查输入格式"
        except ConnectionError:
            return "网络连接失败,请稍后重试"
        except Exception as e:
            return f"工具执行失败: {e}. 请尝试其他方法"

3. 控制执行成本

限制最大迭代次数:

python

agent = Agent(
    llm=llm,
    tools=tools,
    max_iterations=10,  # 最多执行 10 轮
    early_stopping_method="generate"  # 超时后强制生成答案
)

Token 预算管理:

python

class TokenBudgetAgent:
    def __init__(self, max_tokens: int = 10000):
        self.max_tokens = max_tokens
        self.used_tokens = 0
    
    def run(self, query: str):
        while self.used_tokens < self.max_tokens:
            response = self.llm.invoke(query)
            self.used_tokens += count_tokens(response)
            
            if self.is_task_complete(response):
                break
            
            if self.used_tokens >= self.max_tokens * 0.9:
                # 接近预算上限,提醒 Agent 尽快结束
                query = "请在接下来的一轮内给出最终答案"
        
        return response

并行工具调用:

python

import asyncio

async def parallel_tool_execution(tools: list, inputs: list):
    """并行执行多个工具,节省时间"""
    tasks = [tool.arun(inp) for tool, inp in zip(tools, inputs)]
    results = await asyncio.gather(*tasks)
    return results

# 示例: 同时查询多个城市的天气
results = await parallel_tool_execution(
    tools=[WeatherTool()] * 3,
    inputs=["北京", "上海", "深圳"]
)

4. 可观测性与调试

日志记录:

python

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler(f'agent_{datetime.now():%Y%m%d}.log'),
        logging.StreamHandler()
    ]
)

class ObservableAgent:
    def run(self, query: str):
        logging.info(f"[开始] 用户输入: {query}")
        
        for i in range(self.max_iterations):
            logging.info(f"[迭代 {i+1}] 思考中...")
            thought = self.llm.invoke(query)
            logging.info(f"[迭代 {i+1}] Thought: {thought}")
            
            if action := self.parse_action(thought):
                logging.info(f"[迭代 {i+1}] Action: {action['tool']}({action['input']})")
                result = self.execute_tool(action)
                logging.info(f"[迭代 {i+1}] Observation: {result}")
            else:
                logging.info(f"[完成] Final Answer: {thought}")
                break
        
        return thought

中间状态可视化:

python

from rich.console import Console
from rich.table import Table

console = Console()

def visualize_agent_state(state: dict):
    """在终端可视化 Agent 状态"""
    table = Table(title="Agent 执行状态")
    table.add_column("属性", style="cyan")
    table.add_column("值", style="magenta")
    
    table.add_row("当前目标", state.get("goal", ""))
    table.add_row("已完成步骤", str(len(state.get("completed_steps", []))))
    table.add_row("使用的工具", ", ".join(state.get("tools_used", [])))
    table.add_row("Token 消耗", str(state.get("total_tokens", 0)))
    
    console.print(table)

5.1.6 Agent 常见问题与解决方案

问题 1: Agent 陷入循环

现象: Agent 重复执行相同的操作,无法推进任务

原因:

工具返回结果不明确
Prompt 没有引导 Agent 改变策略
缺少循环检测机制

解决方案:

python

class LoopDetector:
    def __init__(self, window_size: int = 3):
        self.action_history = []
        self.window_size = window_size
    
    def is_looping(self, action: str) -> bool:
        """检测是否陷入循环"""
        self.action_history.append(action)
        
        if len(self.action_history) < self.window_size:
            return False
        
        recent = self.action_history[-self.window_size:]
        # 如果最近 N 次操作完全相同,判定为循环
        return len(set(recent)) == 1
    
    def suggest_alternative(self) -> str:
        """建议替代方案"""
        return "检测到重复操作,请尝试其他方法或直接给出当前最佳答案"

# 在 Agent 循环中使用
detector = LoopDetector()
for step in range(max_steps):
    action = agent.decide_action()
    
    if detector.is_looping(action):
        # 强制介入
        agent.add_message(detector.suggest_alternative())
        break
    
    agent.execute(action)

问题 2: 工具调用失败

现象: Agent 调用工具时参数错误或工具不可用

解决方案:

python

class SafeToolExecutor:
    def execute(self, tool_name: str, tool_input: dict) -> str:
        """安全的工具执行器"""
        try:
            # 1. 验证工具存在
            if tool_name not in self.tools:
                return f"错误: 工具 '{tool_name}' 不存在。可用工具: {list(self.tools.keys())}"
            
            tool = self.tools[tool_name]
            
            # 2. 验证参数
            if hasattr(tool, 'args_schema'):
                try:
                    validated = tool.args_schema(**tool_input)
                    tool_input = validated.dict()
                except ValidationError as e:
                    return f"参数错误: {e}. 正确格式: {tool.args_schema.schema()}"
            
            # 3. 执行工具(带超时)
            result = asyncio.wait_for(
                tool.arun(**tool_input),
                timeout=30.0
            )
            return f"执行成功: {result}"
            
        except asyncio.TimeoutError:
            return "工具执行超时(30秒),请尝试简化输入或使用其他工具"
        except Exception as e:
            return f"工具执行失败: {type(e).__name__}: {e}"

问题 3: 输出不稳定

现象: 相同输入,Agent 每次输出结果差异很大

解决方案:

python

# 1. 降低 temperature
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)  # 更确定性

# 2. 使用 Few-shot 示例
AGENT_PROMPT_WITH_EXAMPLES = """
你是一个数据分析 Agent。

示例 1:
用户: 分析 sales.csv 中销量最高的产品
Thought: 我需要先读取文件内容
Action: read_file
Action Input: &#123;&#123;"path": "sales.csv"&#125;&#125;
Observation: [文件内容]
Thought: 现在我可以用 Python 分析数据
Action: python_repl
Action Input: &#123;&#123;"code": "df.groupby('product')['sales'].sum().idxmax()"&#125;&#125;
Observation: Product A
Thought: 我现在知道答案了
Final Answer: 销量最高的产品是 Product A

现在开始处理用户的问题:
{input}
"""

# 3. 添加输出验证
def validate_output(output: str) -> bool:
    """验证输出是否符合预期格式"""
    required_sections = ["Thought", "Action", "Observation", "Final Answer"]
    return all(section in output for section in required_sections)

5.1.7 实战练习

练习 1: 构建天气助手 Agent

需求: 创建一个 Agent,能够查询天气并给出穿衣建议

python

from langchain.agents import initialize_agent, Tool
from langchain_openai import ChatOpenAI

# 定义工具
def get_weather(city: str) -> str:
    """查询天气(模拟)"""
    # 实际应用中调用天气 API
    return f"{city}今天 15°C,多云,有小雨"

weather_tool = Tool(
    name="get_weather",
    func=get_weather,
    description="查询指定城市的实时天气信息。输入: 城市名称(如'北京')"
)

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = initialize_agent(
    tools=[weather_tool],
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 测试
result = agent.run("北京今天天气怎么样?应该穿什么?")
print(result)

预期输出:

> Entering new AgentExecutor chain...
Thought: 我需要先查询北京的天气
Action: get_weather
Action Input: 北京
Observation: 北京今天 15°C,多云,有小雨
Thought: 根据天气情况,我可以给出穿衣建议了
Final Answer: 北京今天 15°C,多云有小雨。建议穿着:
- 外套: 薄款防水外套或风衣
- 内搭: 长袖衬衫或薄毛衣
- 裤子: 长裤
- 鞋子: 防水鞋或雨靴
- 记得带伞!

练习 2: 多工具协作 Agent

需求: 创建一个能搜索信息并进行数学计算的 Agent

python

from langchain.agents import load_tools, initialize_agent

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 加载内置工具
tools = load_tools(
    ["serpapi", "llm-math"],  # 搜索工具 + 数学工具
    llm=llm
)

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)

# 测试复杂任务
result = agent.run(
    "特斯拉 Model 3 的续航里程是多少公里?如果每天开 50 公里,能用多少天?"
)

练习 3: 使用 LangGraph 构建可控 Agent

需求: 实现一个带条件分支的客服 Agent

python

from langgraph.graph import StateGraph, END
from typing import TypedDict

class CustomerServiceState(TypedDict):
    query: str
    category: str  # "退款" | "技术支持" | "其他"
    response: str

def classify_query(state: CustomerServiceState):
    """分类用户问题"""
    query = state["query"]
    if "退款" in query or "退货" in query:
        return {"category": "退款"}
    elif "无法使用" in query or "报错" in query:
        return {"category": "技术支持"}
    else:
        return {"category": "其他"}

def handle_refund(state: CustomerServiceState):
    """处理退款问题"""
    return {"response": "退款流程: 1. 登录账户 2. 进入订单页面 3. 点击申请退款"}

def handle_tech_support(state: CustomerServiceState):
    """处理技术问题"""
    return {"response": "请提供错误截图,我们的技术团队会在 24 小时内回复"}

def handle_other(state: CustomerServiceState):
    """处理其他问题"""
    return {"response": "您的问题已转接人工客服,请稍候"}

# 构建图
graph = StateGraph(CustomerServiceState)
graph.add_node("classify", classify_query)
graph.add_node("refund", handle_refund)
graph.add_node("tech", handle_tech_support)
graph.add_node("other", handle_other)

graph.set_entry_point("classify")
graph.add_conditional_edges(
    "classify",
    lambda s: s["category"],
    {"退款": "refund", "技术支持": "tech", "其他": "other"}
)
graph.add_edge("refund", END)
graph.add_edge("tech", END)
graph.add_edge("other", END)

app = graph.compile()

# 测试
result = app.invoke({"query": "我要退款"})
print(result["response"])

5.1.8 学习资源

推荐阅读:

ReAct 论文 - Reasoning and Acting 原理
LangGraph 文档 - 官方教程
AutoGPT GitHub - 自主 Agent 实现
CrewAI 文档 - 多 Agent 协作框架

实战项目:

LangChain Agent 示例库
Awesome LLM Agents - Agent 资源汇总

进阶主题:

下一节 5.2: 工具调用与 Function Calling
下一节 5.3: Agent 规划与推理策略
下一节 5.4: 多 Agent 协作系统

关键要点总结:

✅ Agent = LLM + Tools + Memory + Planning
✅ ReAct 是最实用的 Agent 架构
✅ 工具设计要清晰、单一职责、错误处理完善
✅ 控制执行成本: 限制迭代次数、Token 预算
✅ 可观测性: 日志、状态可视化、循环检测
✅ 框架选择: 简单用 LangChain,复杂用 LangGraph

5.1.1 什么是 Agent? ​

定义与特征 ​

Agent vs 传统编程 ​

5.1.2 Agent 核心架构 ​

ReAct 模式 (Reasoning + Acting) ​

核心组件详解 ​

5.1.3 Agent 类型与应用场景 ​

按能力分类 ​

按应用领域分类 ​

5.1.4 Agent 开发框架对比 ​

主流框架选型 ​

LangGraph 快速入门 ​

5.1.5 Agent 设计最佳实践 ​

1. Prompt 工程 ​

2. 工具设计原则 ​

3. 控制执行成本 ​

4. 可观测性与调试 ​

5.1.6 Agent 常见问题与解决方案 ​

问题 1: Agent 陷入循环 ​

问题 2: 工具调用失败 ​

问题 3: 输出不稳定 ​

5.1.7 实战练习 ​

练习 1: 构建天气助手 Agent ​

练习 2: 多工具协作 Agent ​

练习 3: 使用 LangGraph 构建可控 Agent ​

5.1.8 学习资源 ​

5.1.1 什么是 Agent?

定义与特征

Agent vs 传统编程

5.1.2 Agent 核心架构

ReAct 模式 (Reasoning + Acting)

核心组件详解

5.1.3 Agent 类型与应用场景

按能力分类

按应用领域分类

5.1.4 Agent 开发框架对比

主流框架选型

LangGraph 快速入门

5.1.5 Agent 设计最佳实践

1. Prompt 工程

2. 工具设计原则

3. 控制执行成本

4. 可观测性与调试

5.1.6 Agent 常见问题与解决方案

问题 1: Agent 陷入循环

问题 2: 工具调用失败

问题 3: 输出不稳定

5.1.7 实战练习

练习 1: 构建天气助手 Agent

练习 2: 多工具协作 Agent

练习 3: 使用 LangGraph 构建可控 Agent

5.1.8 学习资源