第四部分:发展路线与前沿

4.1 第一代:ReAct Agent

ReAct(Reasoning + Acting)是 2022 年底由 Shunyu Yao 等人在论文《ReAct: Synergizing Reasoning and Acting in Language Models》中提出的范式。

核心思想:

在 Agent 的推理过程中,交替输出「思考」和「行动」。思考是为了推理下一步做什么,行动是为了获取更多信息或执行操作。

原始论文中的 ReAct 轨迹:

思考 1: 我需要知道 Faraday 的出生年份才能回答这个问题。
行动 1: Search[Faraday 出生年份]
观察 1: 迈克尔·法拉第(1791年9月22日-1867年8月25日)
思考 2: 法拉第出生于1791年。我需要计算他多少岁时发现了电磁感应(1831年)。
行动 2: Calculate[1831 - 1791]
观察 2: 40
思考 3: 法拉第在40岁时发现了电磁感应。
回答: 法拉第在40岁时发现了电磁感应现象。

ReAct 的实现原理:

ReAct 的特殊之处在于,它把「思考」和「行动」放在同一个格式中。LLM 的输出里交替包含:

思考(Thought): Agent 在当前状态下的推理过程
行动(Action): Agent 决定执行的工具调用
观察(Observation): 工具调用的结果(来自环境)

这个格式让 LLM 能够追踪自己的推理链条,不容易迷失。

# ReAct Agent 的完整实现
class ReActAgent:
    def __init__(self, llm, tools: List[Tool], max_steps: int = 20):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps
        
        self.react_prompt = """你是一个能推理和行动的 AI 助手。
可用工具:
{tool_descriptions}

请按照以下格式输出:
思考:<你的推理过程>
行动:<工具名称>(参数)
观察:<工具返回的结果>
...(重复)
回答:<最终答案>

用户:{question}"""
    
    def run(self, question: str) -> str:
        prompt = self.react_prompt.format(
            tool_descriptions=self._format_tools(),
            question=question
        )
        
        history = [{"role": "user", "content": prompt}]
        
        for step in range(self.max_steps):
            response = self.llm.chat(history)
            
            # 解析 LLM 输出
            action = self._parse_action(response.content)
            
            if action["type"] == "answer":
                return action["content"]
            
            elif action["type"] == "tool_call":
                # 执行工具
                tool_name = action["name"]
                tool_args = action["arguments"]
                result = self.tools[tool_name].fn(**tool_args)
                
                # 把结果放回对话
                history.append({"role": "assistant", "content": response.content})
                history.append({"role": "user", "content": f"观察:{result}"})
        
        return "达到最大步数限制"
    
    def _parse_action(self, text: str) -> dict:
        """从 LLM 输出中解析行动"""
        # 检查是否包含"回答:"
        if "回答:" in text:
            answer = text.split("回答:")[-1].strip()
            return {"type": "answer", "content": answer}
        
        # 检查是否包含"行动:"
        if "行动:" in text:
            action_line = [l for l in text.split("\n") if l.startswith("行动:")][0]
            action_text = action_line.replace("行动:", "").strip()
            # 解析工具名和参数(格式:ToolName(arg1, arg2))
            import re
            match = re.match(r"(\w+)\((.+)\)", action_text)
            if match:
                return {
                    "type": "tool_call",
                    "name": match.group(1),
                    "arguments": self._parse_args(match.group(2))
                }
        
        return {"type": "answer", "content": text}

ReAct 的优缺点:

优点:

  • 推理过程透明(每一步都能看到 Agent 在想什么)
  • 灵活,能适应各种突发情况
  • 原理简单,容易实现

缺点:

  • 效率低(每一步都要 LLM 推理)
  • 上下文快速膨胀
  • 没有结构化规划,容易陷入局部最优

ReAct 是第一代 Agent 的基石。今天几乎所有 Agent 框架都或多或少用了 ReAct 的思想。


4.2 第二代:多 Agent 协作

第二代 Agent 的核心创新是:不再用一个大模型处理所有事,而是用多个专业化的 Agent 组成团队。

驱动因素:

  1. 性能瓶颈:单 Agent 受限于上下文窗口,复杂任务容易丢信息
  2. 角色冲突:同一个 Agent 既要写代码又要测试又要部署,提示词相互打架
  3. 效率问题:全能的 Agent 在每个领域都不如专家 Agent

多 Agent 协作的关键技术:

1. 任务分解(Task Decomposition):
   把复杂任务拆成独立可执行的子任务

2. Agent 路由(Agent Routing):
   把子任务分配给最适合的 Agent

3. 上下文隔离(Context Isolation):
   每个 Agent 只有自己需要的上下文

4. 信息聚合(Information Aggregation):
   把多个 Agent 的结果合并成完整输出

代表项目:

  • AutoGen(Microsoft):允许多个 Agent 通过对话协作
  • CrewAI:轻量级多 Agent 框架
  • LangGraph:有向图多 Agent 编排
  • Hermes Agent 多 Agent 模式:Orchestrator + Worker
# CrewAI 风格的多 Agent 示例
from crewai import Agent, Task, Crew

# 定义 Agent
researcher = Agent(
    name="研究员",
    goal="搜索和收集信息",
    backstory="你是一个顶级研究员,擅长从海量信息中找到关键内容",
    tools=[web_search, read_document]
)

writer = Agent(
    name="写手",
    goal="写出高质量的文章",
    backstory="你是一个资深技术写手,擅长把复杂概念讲清楚",
    tools=[write_file]
)

reviewer = Agent(
    name="审核员",
    goal="确保输出内容的质量",
    backstory="你是一个严格的审核员,不会放过任何错误",
    tools=[read_file, lint_check]
)

# 定义任务
research_task = Task(
    description="研究 AI Agent 最新发展",
    agent=researcher
)

write_task = Task(
    description="基于研究结果写一篇文章",
    agent=writer,
    context=[research_task]  # 依赖研究结果
)

review_task = Task(
    description="审核文章质量",
    agent=reviewer,
    context=[write_task]  # 依赖文章
)

# 组建团队
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    verbose=True
)

# 执行
result = crew.kickoff()

4.3 第三代:垂直 Agent / 工具型 Agent

第三代 Agent 放弃了「通用」路线,转向「专精」。每个 Agent 只做一件事,但做到极致。

垂直 Agent 的设计哲学:

通用 Agent = 什么都会一点,但什么都不精通 垂直 Agent = 一件事做到 95 分,其他事不会

典型例子:

Claude Code(Anthropic 的编码 Agent)

  • 只做一件事:写代码
  • 工具集:文件操作 + 终端执行 + Git 操作
  • 深度集成了项目上下文理解

Codex CLI(OpenAI 的开源编码 Agent)

  • 只做一件事:在终端里写代码
  • 纯终端交互,没有 GUI
  • 特点是「需求 → 可运行代码」的转换效率极高

各种垂直 Agent:

  • 数据分析 Agent:SQL + Python + 可视化,只做数据分析
  • 运维 Agent:SSH + K8s + 监控,只做运维
  • 客服 Agent:CRM + 知识库 + 工单系统,只做客服
  • 法律 Agent:法条检索 + 合同审查,只做法律

垂直 Agent 的核心优势:

  1. 工具集精简:只配需要的工具,LLM 不会选择困难
  2. Prompt 优化:System Prompt 高度定制,不需要兼容多种任务
  3. 上下文高效:不需要在多个任务间切换,上下文利用率高
  4. 准确率高:专精任务的表现远好于通用 Agent
# 垂直 Agent 示例:数据分析 Agent
class DataAnalystAgent:
    def __init__(self, llm):
        self.llm = llm
        self.tools = {
            "query_database": Tool(
                name="query_database",
                description="执行 SQL 查询,从数据库中获取数据",
                parameters={
                    "sql": {"type": "string", "description": "SQL 查询语句"}
                },
                fn=self._execute_sql
            ),
            "analyze_data": Tool(
                name="analyze_data",
                description="对数据进行统计分析,返回统计结果",
                parameters={
                    "data": {"type": "string", "description": "要分析的数据(CSV 格式)"},
                    "analysis_type": {"type": "string", "description": "分析类型:describe/correlation/regression"}
                },
                fn=self._analyze
            ),
            "create_chart": Tool(
                name="create_chart",
                description="生成数据可视化图表",
                parameters={
                    "data": {"type": "string", "description": "数据(CSV 格式)"},
                    "chart_type": {"type": "string", "description": "图表类型"},
                    "title": {"type": "string", "description": "图表标题"}
                },
                fn=self._visualize
            ),
            "write_report": Tool(
                name="write_report",
                description="生成数据分析报告",
                parameters={
                    "content": {"type": "string", "description": "报告内容"}
                },
                fn=self._generate_report
            )
        }
        
        self.system_prompt = """你是数据分析专家 Agent。
你的唯一任务是根据用户的描述,从数据库中提取数据、分析、可视化并生成报告。

工作流程:
1. 理解用户需求
2. 查询数据库获取数据
3. 进行统计分析
4. 生成可视化图表
5. 撰写分析报告

你只做数据分析相关的事。如果用户问无关问题,礼貌拒绝。"""
    
    def run(self, query: str) -> str:
        """处理数据分析请求"""
        # ...标准 Agent 循环
        pass

垂直 Agent 的生态现状:

领域 代表产品 核心能力 工具集
编码 Claude Code, Codex CLI 代码生成、调试、重构 文件、终端、Git
数据 Julius AI, Dataherald SQL、分析、可视化 数据库、Python
客服 Intercom Fin, Zendesk AI 工单回复、知识库 CRM、知识库
销售 Gong, Clari 客户分析、预测 CRM、邮件
运维 内部工具 监控、告警、修复 K8s、SSH、监控

4.4 第四代:自主 Agent(未来方向)

第四代 Agent 还在早期阶段,但几个关键趋势已经清晰。

自主 Agent 的核心特征:

  1. 长期目标追踪

    • 不只是完成当前任务,而是理解长期目标
    • 能在多次 session 间持续追踪进度
    • 知道「今天做到哪了,明天接着干」
  2. 主动学习能力

    • 遇到不认识的工具,会自己看文档学习怎么用
    • 遇到不熟悉的领域,会自己查资料补充知识
    • 从错误中学习,不再犯同样的错
  3. 自我反思与改进

    • 每次任务完成后自动总结经验
    • 识别自己的不足之处
    • 优化自己的工作流程
  4. Agent 互操作性

    • 不同厂商的 Agent 可以互相调用
    • 一个 Agent 可以请求另一个 Agent 的帮助
    • 标准化的通信协议

MCP 协议(Model Context Protocol)

Anthropic 提出的 MCP 协议,是 Agent 领域的「USB 标准」。它定义了一个标准化的方式,让 LLM 和外部工具/数据源通信。

MCP 的核心概念:

  • Host:LLM 应用(如 Claude Desktop)
  • Client:与 MCP 服务器建立 1:1 连接的客户端
  • Server:暴露特定资源、工具和提示的轻量级程序
# MCP 协议的抽象工作方式
# 一个 MCP Server 的定义
class WeatherMCPServer:
    """MCP 服务器:提供天气查询功能"""
    
    @mcp.tool
    def get_weather(self, city: str) -> str:
        """获取城市天气"""
        data = call_weather_api(city)
        return json.dumps(data)
    
    @mcp.tool
    def get_forecast(self, city: str, days: int) -> str:
        """获取天气预报"""
        data = call_forecast_api(city, days)
        return json.dumps(data)

# MCP 服务器注册到 Agent
mcp_server = MCPServer("weather", WeatherMCPServer())
agent.register_mcp(mcp_server)

# Agent 现在可以用 MCP 服务器提供的工具
agent.run("北京明天会下雨吗?")
# → Agent 调用 weather 服务器的 get_forecast 工具

A2A 协议(Agent-to-Agent)

Google 提出的 A2A 协议,解决的是 Agent 之间的通信问题。

A2A 的核心概念:

  • Agent Card:Agent 的「名片」,描述它能做什么
  • Task:一个 Agent 发给另一个 Agent 的工作请求
  • Artifact:Agent 完成任务后产出的成果
  • Message:Agent 间的通信消息
# A2A 协议的概念示例
class AgentCard:
    """Agent 的名片"""
    name: str
    description: str
    capabilities: List[str]  # 能力列表
    endpoints: List[str]     # 通信端点
    auth_required: bool

# Agent A 发现 Agent B
agent_b_card = discover_agent("agent_b")
# 输出:
# AgentCard(
#   name="数据分析师",
#   description="擅长数据分析和可视化",
#   capabilities=["sql查询", "统计分析", "图表生成"],
#   endpoints=["http://agent-b.internal:8080/a2a"],
#   auth_required=True
# )

# Agent A 向 Agent B 发送任务
task_request = A2ATask(
    task_id="task_001",
    type="data_analysis",
    input={"sql": "SELECT * FROM sales WHERE date > '2024-01-01'"},
    callback_url="http://agent-a.internal:8080/callback"
)

# Agent B 异步处理,完成后回调
response = await send_a2a_task("http://agent-b.internal:8080/a2a", task_request)

第四代的现实进展:

技术 状态 代表项目
MCP 协议 正在推广 Claude Desktop, Cursor 已支持
A2A 协议 标准制定中 Google 提出,社区讨论阶段
自主反思 实验阶段 Reflexion 论文, Voyager
长期记忆 生产可用 Mem0, 各种 Agent 框架
主动学习 研究阶段 少数实验项目
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐