AI Agent 开发实战：从 ReAct 到 Multi-Agent 系统设计

什么是 AI Agent

AI Agent（智能代理）是能够自主感知环境、做出决策并执行行动以完成目标的系统。与传统的对话式 AI 不同，Agent 具备以下核心能力：

• 自主规划：将复杂任务拆解为可执行的子步骤
• 工具调用：调用外部 API、数据库、搜索引擎等
• 记忆系统：存储和检索上下文信息
• 反思修正：根据执行结果调整策略

ReAct 模式：Agent 的基础范式

ReAct（Reasoning + Acting）是 Agent 最经典的推理模式，交替进行”思考”和”行动”：

思考：用户想要查询北京明天的天气，我需要调用天气 API
行动：调用 weather_api(city="北京", date="tomorrow")
观察：北京明天晴，最高温 32°C，最低温 22°C
思考：我已经获得了天气信息，可以回答用户了
回答：北京明天天气晴朗，最高温度 32°C，最低温度 22°C，适合外出。

LangChain ReAct Agent 实现

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

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 实际实现调用搜索 API
    return f"搜索结果：{query} 的相关信息..."

@tool
def calculate(expression: str) -> str:
    """数学计算"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

@tool
def write_file(filename: str, content: str) -> str:
    """将内容写入文件"""
    with open(filename, 'w') as f:
        f.write(content)
    return f"已写入 {filename}"

tools = [search_web, calculate, write_file]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(llm, tools, prompt=None)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行任务
result = executor.invoke({
    "input": "搜索 Python 3.13 的新特性，总结要点并保存到 python313.txt"
})

工具调用（Tool Use）详解

自定义工具定义

from pydantic import BaseModel, Field
from langchain.tools import StructuredTool

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    date: str = Field(description="日期，格式 YYYY-MM-DD")

def get_weather(city: str, date: str) -> dict:
    """获取指定城市和日期的天气预报"""
    # 调用天气 API
    return {"city": city, "date": date, "weather": "晴", "temp": 25}

weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="weather",
    description="获取天气预报",
    args_schema=WeatherInput
)

Function Calling vs ReAct

特性	Function Calling	ReAct
推理方式	模型直接决定调用	文本推理后调用
可观测性	较低	高（可见思考过程）
速度	快	较慢（多轮推理）
准确性	高（原生支持）	依赖 Prompt
适用模型	GPT/Claude 等	任何模型

记忆系统（Memory）

短期记忆

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    max_token_limit=2000  # 限制上下文长度
)

长期记忆：向量数据库

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 存储对话历史到向量数据库
vectorstore = Chroma(
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./agent_memory"
)

# 检索相关记忆
def recall(query: str, k: int = 3):
    docs = vectorstore.similarity_search(query, k=k)
    return [doc.page_content for doc in docs]

记忆层级

层级	存储方式	容量	检索速度	用途
工作记忆	上下文窗口	小	即时	当前对话
短期记忆	缓冲区	中	快	近期交互
长期记忆	向量数据库	大	中	历史知识
结构化记忆	数据库	大	快	用户画像

Multi-Agent 多智能体系统

AutoGen 框架

import autogen

config_list = [{
    "model": "gpt-4o",
    "api_key": "your-api-key"
}]

# 创建用户代理
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10
)

# 创建研究员 Agent
researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="你是一个研究助手，负责搜索和整理信息。",
    llm_config={"config_list": config_list}
)

# 创建写手 Agent
writer = autogen.AssistantAgent(
    name="Writer",
    system_message="你是一个专业写手，将研究结果整理成文章。",
    llm_config={"config_list": config_list}
)

# 创建审查员 Agent
reviewer = autogen.AssistantAgent(
    name="Reviewer",
    system_message="你是内容审查员，检查文章质量和准确性。",
    llm_config={"config_list": config_list}
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, writer, reviewer],
    messages=[],
    max_round=15
)

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

# 启动任务
user_proxy.initiate_chat(
    manager,
    message="写一篇关于 AI Agent 发展趋势的技术文章"
)

CrewAI 团队编排

from crewai import Agent, Task, Crew

researcher = Agent(
    role='研究员',
    goal='收集和整理 AI Agent 相关的最新信息',
    backstory='你是一位资深技术研究员，擅长快速收集和总结信息。',
    tools=[search_tool]
)

writer = Agent(
    role='技术写手',
    goal='将研究结果转化为清晰易懂的文章',
    backstory='你是一位经验丰富的技术写手，擅长将复杂概念简化。'
)

research_task = Task(
    description='研究 AI Agent 的最新发展趋势，包括 ReAct、Multi-Agent 等',
    agent=researcher,
    expected_output='一份包含关键发现的研究报告'
)

writing_task = Task(
    description='基于研究报告撰写一篇 2000 字的技术文章',
    agent=writer,
    expected_output='一篇完整的中文技术文章',
    context=[research_task]  # 依赖研究任务的输出
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential
)

result = crew.kickoff()

Agent 架构设计模式

模式	描述	适用场景
单 Agent	一个 Agent 完成所有任务	简单任务、快速原型
顺序流水线	多个 Agent 依次处理	内容创作、数据处理
并行协作	多个 Agent 同时工作	多角度分析、投票决策
层级管理	主管 Agent 分配任务	复杂项目管理
竞争选择	多个 Agent 各自完成，选最优	创意生成、方案比选

实战案例：自动研究助手

def research_assistant(topic: str):
    """自动研究助手：搜索→分析→总结→生成报告"""
    
    # 1. 规划研究步骤
    plan = agent.invoke(f"为主题'{topic}'制定研究计划")
    
    # 2. 并行搜索多个方面
    with ThreadPoolExecutor() as executor:
        futures = [executor.invoke(search_tool, q) for q in plan.questions]
        results = [f.result() for f in futures]
    
    # 3. 分析和综合
    analysis = agent.invoke(f"分析以下信息：{results}")
    
    # 4. 生成报告
    report = agent.invoke(f"基于分析结果撰写研究报告")
    
    # 5. 自动保存
    write_file(f"report_{topic}.md", report)
    
    return report

总结

AI Agent 是大模型从”对话工具”走向”生产力工具”的关键一步。从单 Agent 的 ReAct 推理到 Multi-Agent 的协作系统，Agent 的能力边界正在不断扩展。掌握 Agent 开发，意味着能让 AI 自主完成复杂的多步骤任务，这将是未来软件开发的重要范式。