前言：从"工具"到"助手"

大模型的进化路径：

传统 AI：你说一句，它做一步，完全被动。

用户: 帮我查一下北京的天气
AI: 北京今天晴，25度。（结束）

用户: 那我应该穿什么？
AI: 建议穿薄外套。（结束）

用户: 帮我把这个加到日程里
AI: 好的，已添加。（结束）

Agent：你说目标，它自己规划、执行、反思。

用户: 我明天要去北京出差，帮我准备一下

Agent 思考: 用户要去北京出差，我需要：
  1. 查询北京明天的天气
  2. 根据天气推荐穿着
  3. 查看用户的日程安排
  4. 提醒可能的冲突
  5. 生成出差清单
  
Agent 执行:
  → 调用天气 API: 北京明天多云，18-24度
  → 调用日历 API: 明天下午有个会议需要改期
  → 生成结果...
  
Agent: 我帮您做了以下准备：
  1. 天气：明天北京多云，18-24度，建议带薄外套
  2. 日程冲突：您明天下午3点有个本地会议，需要改期吗？
  3. 出差清单：身份证、充电器、笔记本...
  需要我帮您处理会议改期吗？

一、什么是 Agent？

1.1 Agent 的定义

Agent = LLM + 规划能力 + 工具使用 + 记忆

1.2 Agent vs 传统 Chatbot

1.3 Agent 的核心循环

这就是著名的 ReAct（Reasoning + Acting） 模式：

• Reasoning：思考下一步该做什么
• Acting：执行动作
• Observation：观察结果
• 循环直到完成目标

二、ReAct：Agent 的基础范式

2.1 ReAct 的核心思想

让 LLM 在回答问题时，交替进行"思考"和"行动"：

问题: 《流浪地球》的导演出生于哪一年？

思考1: 我需要先找到《流浪地球》的导演是谁
行动1: search("流浪地球 导演")
观察1: 《流浪地球》的导演是郭帆

思考2: 现在我知道导演是郭帆，需要查他的出生年份
行动2: search("郭帆 出生年份")
观察2: 郭帆，1980年出生于山东济宁

思考3: 我已经得到答案了
答案: 1980年

2.2 ReAct Prompt 模板

REACT_PROMPT = """你是一个智能助手，可以使用以下工具来回答问题：

{tools_description}

请按照以下格式回答问题：

问题: 用户的问题
思考: 分析问题，决定下一步行动
行动: 工具名称[参数]
观察: 工具返回的结果
... (思考/行动/观察 可以重复多次)
思考: 我已经得到了最终答案
答案: 最终答案

注意：
1. 每次只执行一个行动
2. 根据观察结果决定下一步
3. 如果信息足够，直接给出答案

开始！

问题: {question}
思考:"""


def run_react_agent(question: str, tools: dict, max_iterations: int = 5):
    """运行 ReAct Agent"""
    
    # 构建工具描述
    tools_desc = "\n".join([
        f"- {name}: {tool['description']}"
        for name, tool in tools.items()
    ])
    
    prompt = REACT_PROMPT.format(
        tools_description=tools_desc,
        question=question,
    )
    
    for i in range(max_iterations):
        # LLM 生成下一步
        response = llm.generate(prompt, stop=["观察:"])
        prompt += response
        
        # 解析行动
        if "答案:" in response:
            # 得到最终答案
            answer = response.split("答案:")[-1].strip()
            return answer
        
        if "行动:" in response:
            # 执行行动
            action_line = [l for l in response.split("\n") if "行动:" in l][0]
            action = parse_action(action_line)
            
            # 调用工具
            tool_name, tool_input = action
            observation = tools[tool_name]["function"](tool_input)
            
            # 添加观察结果
            prompt += f"\n观察: {observation}\n思考:"
    
    return "达到最大迭代次数，未能得出答案"


def parse_action(action_line: str) -> tuple:
    """解析行动，如 '行动: search[流浪地球导演]' """
    import re
    match = re.search(r'行动:\s*(\w+)\[(.*?)\]', action_line)
    if match:
        return match.group(1), match.group(2)
    return None, None

2.3 ReAct 完整示例

# 定义工具
tools = {
    "search": {
        "description": "搜索信息，输入搜索关键词",
        "function": lambda q: web_search(q),
    },
    "calculator": {
        "description": "计算数学表达式，输入数学表达式",
        "function": lambda expr: eval(expr),
    },
    "weather": {
        "description": "查询天气，输入城市名",
        "function": lambda city: get_weather(city),
    },
}

# 运行
result = run_react_agent(
    question="北京今天的温度是多少华氏度？",
    tools=tools,
)

# Agent 的思考过程：
# 思考: 我需要先查询北京的天气获取摄氏温度，然后转换为华氏温度
# 行动: weather[北京]
# 观察: 北京今天晴，气温 25°C
# 思考: 现在我知道是25°C，需要转换为华氏度。公式是 F = C * 9/5 + 32
# 行动: calculator[25 * 9/5 + 32]
# 观察: 77.0
# 思考: 我已经得到了答案
# 答案: 北京今天的温度是 77°F（华氏度）

三、Function Calling：工具调用的标准方式

3.1 什么是 Function Calling？

Function Calling 是 OpenAI 提出的标准化工具调用方式，现在大多数 LLM 都支持。

3.2 OpenAI Function Calling

from openai import OpenAI

client = OpenAI()

# 定义工具（函数）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'、'上海'",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位",
                    },
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索网页获取信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词",
                    },
                },
                "required": ["query"],
            },
        },
    },
]


def get_weather(city: str, unit: str = "celsius") -> str:
    """模拟天气 API"""
    # 实际应该调用真实的天气 API
    return f"{city}今天晴，气温 25°C，空气质量良好"


def search_web(query: str) -> str:
    """模拟搜索 API"""
    return f"搜索结果：关于'{query}'的相关信息..."


# 工具函数映射
available_functions = {
    "get_weather": get_weather,
    "search_web": search_web,
}


def run_with_tools(user_message: str):
    """带工具调用的对话"""
    
    messages = [{"role": "user", "content": user_message}]
    
    # 第一次调用：让 LLM 决定是否需要使用工具
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools,
        tool_choice="auto",  # 自动决定是否使用工具
    )
    
    response_message = response.choices[0].message
    
    # 检查是否需要调用工具
    if response_message.tool_calls:
        # 处理所有工具调用
        messages.append(response_message)
        
        for tool_call in response_message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)
            
            # 执行函数
            function_response = available_functions[function_name](**function_args)
            
            # 将结果添加到消息
            messages.append({
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": function_response,
            })
        
        # 第二次调用：让 LLM 根据工具结果生成回答
        final_response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
        )
        
        return final_response.choices[0].message.content
    
    else:
        # 不需要工具，直接返回
        return response_message.content


# 使用示例
result = run_with_tools("北京今天天气怎么样？适合户外运动吗？")
print(result)

3.3 并行工具调用

GPT-4 支持一次返回多个工具调用：

def run_parallel_tools(user_message: str):
    """并行工具调用"""
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}],
        tools=tools,
        tool_choice="auto",
    )
    
    tool_calls = response.choices[0].message.tool_calls
    
    if tool_calls:
        # 并行执行所有工具（实际可以用 asyncio）
        import concurrent.futures
        
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {}
            for tool_call in tool_calls:
                fn_name = tool_call.function.name
                fn_args = json.loads(tool_call.function.arguments)
                future = executor.submit(available_functions[fn_name], **fn_args)
                futures[tool_call.id] = future
            
            # 收集结果
            results = {
                call_id: future.result()
                for call_id, future in futures.items()
            }
        
        return results


# 示例：一次查询多个城市的天气
result = run_parallel_tools("北京和上海今天的天气分别怎么样？")
# LLM 会返回两个工具调用：get_weather("北京") 和 get_weather("上海")

四、常见 Agent 架构

4.1 Agent 架构全景

4.2 Plan-and-Execute

先制定完整计划，再逐步执行：

class PlanAndExecuteAgent:
    """Plan-and-Execute Agent"""
    
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm
        self.executor = executor_llm
        self.tools = tools
    
    def plan(self, goal: str) -> List[str]:
        """制定计划"""
        prompt = f"""为完成以下目标，制定一个详细的步骤计划。
        
目标：{goal}

请以编号列表的形式输出计划，每个步骤应该是具体、可执行的：
1. 
2. 
..."""
        
        response = self.planner.generate(prompt)
        steps = self.parse_plan(response)
        return steps
    
    def execute_step(self, step: str, context: str) -> str:
        """执行单个步骤"""
        prompt = f"""执行以下步骤。

已有上下文：
{context}

当前步骤：{step}

可用工具：{self.tools_description}

请执行这个步骤并返回结果。"""
        
        return self.executor.generate(prompt)
    
    def run(self, goal: str) -> str:
        """运行 Agent"""
        # 1. 制定计划
        plan = self.plan(goal)
        print(f"计划：{plan}")
        
        # 2. 逐步执行
        context = ""
        for i, step in enumerate(plan):
            print(f"执行步骤 {i+1}: {step}")
            result = self.execute_step(step, context)
            context += f"\n步骤{i+1}结果: {result}"
        
        # 3. 总结
        return self.summarize(goal, context)

4.3 Multi-Agent：多智能体协作

多个 Agent 协作完成任务：

class MultiAgentSystem:
    """多 Agent 系统"""
    
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "critic": CriticAgent(),
        }
        self.manager = ManagerAgent()
    
    def run(self, task: str) -> str:
        """运行多 Agent 协作"""
        
        # Manager 分析任务，分配子任务
        subtasks = self.manager.decompose(task)
        
        results = {}
        for subtask in subtasks:
            agent_name = subtask["agent"]
            agent_task = subtask["task"]
            
            # 执行子任务
            result = self.agents[agent_name].run(agent_task)
            results[agent_name] = result
        
        # Manager 整合结果
        final_output = self.manager.synthesize(results)
        
        return final_output


class ManagerAgent:
    """管理 Agent"""
    
    def decompose(self, task: str) -> List[dict]:
        """分解任务"""
        prompt = f"""将以下任务分解为子任务，并分配给合适的 Agent。

可用 Agent：
- researcher: 负责搜索和收集信息
- writer: 负责撰写内容
- critic: 负责审核和改进

任务：{task}

输出格式（JSON）：
[{{"agent": "agent_name", "task": "子任务描述"}}]"""
        
        response = self.llm.generate(prompt)
        return json.loads(response)
    
    def synthesize(self, results: dict) -> str:
        """整合结果"""
        prompt = f"""基于各 Agent 的工作结果，生成最终输出。

各 Agent 结果：
{json.dumps(results, ensure_ascii=False, indent=2)}

请整合以上结果，生成完整、连贯的最终输出："""
        
        return self.llm.generate(prompt)

4.4 AutoGen：微软的多 Agent 框架

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager


def build_autogen_system():
    """使用 AutoGen 构建多 Agent 系统"""
    
    # 配置 LLM
    config_list = [{"model": "gpt-4", "api_key": "..."}]
    
    # 创建 Agent
    assistant = AssistantAgent(
        name="Assistant",
        llm_config={"config_list": config_list},
        system_message="你是一个有帮助的 AI 助手。",
    )
    
    coder = AssistantAgent(
        name="Coder",
        llm_config={"config_list": config_list},
        system_message="你是一个专业的程序员，负责编写代码。",
    )
    
    critic = AssistantAgent(
        name="Critic",
        llm_config={"config_list": config_list},
        system_message="你是一个代码审查专家，负责审查代码质量。",
    )
    
    # 用户代理（可以执行代码）
    user_proxy = UserProxyAgent(
        name="User",
        human_input_mode="NEVER",
        code_execution_config={"work_dir": "coding"},
    )
    
    # 创建群聊
    groupchat = GroupChat(
        agents=[user_proxy, assistant, coder, critic],
        messages=[],
        max_round=10,
    )
    
    manager = GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list})
    
    # 启动对话
    user_proxy.initiate_chat(
        manager,
        message="写一个 Python 函数来计算斐波那契数列",
    )


# Agent 对话示例：
# User: 写一个 Python 函数来计算斐波那契数列
# Coder: 好的，这是一个计算斐波那契数列的函数...
# Critic: 代码可以工作，但建议添加缓存优化...
# Coder: 好的，这是优化后的版本...
# Assistant: 总结：我们完成了一个优化的斐波那契函数...

五、使用 LangChain 构建 Agent

5.1 LangChain Agent 快速入门

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


def build_langchain_agent():
    """使用 LangChain 构建 Agent"""
    
    # 创建 LLM
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    # 定义工具
    tools = [
        Tool(
            name="Search",
            func=lambda q: search_web(q),
            description="搜索网页信息。输入应该是搜索关键词。",
        ),
        Tool(
            name="Calculator",
            func=lambda expr: eval(expr),
            description="计算数学表达式。输入应该是有效的数学表达式。",
        ),
        Tool(
            name="Weather",
            func=lambda city: get_weather(city),
            description="查询城市天气。输入应该是城市名称。",
        ),
    ]
    
    # 获取 ReAct prompt
    prompt = hub.pull("hwchase17/react")
    
    # 创建 Agent
    agent = create_react_agent(llm, tools, prompt)
    
    # 创建 Agent 执行器
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        handle_parsing_errors=True,
        max_iterations=5,
    )
    
    return agent_executor


# 使用
agent = build_langchain_agent()
result = agent.invoke({"input": "北京今天温度是多少华氏度？"})
print(result["output"])

5.2 自定义工具

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional


class WeatherInput(BaseModel):
    """天气查询的输入参数"""
    city: str = Field(description="城市名称")
    date: Optional[str] = Field(default="today", description="日期，如 'today', 'tomorrow'")


class WeatherTool(BaseTool):
    """自定义天气工具"""
    
    name: str = "weather"
    description: str = "查询指定城市的天气信息"
    args_schema: Type[BaseModel] = WeatherInput
    
    def _run(self, city: str, date: str = "today") -> str:
        """同步执行"""
        # 调用天气 API
        return f"{city} {date} 的天气是晴天，温度 25°C"
    
    async def _arun(self, city: str, date: str = "today") -> str:
        """异步执行"""
        return self._run(city, date)


class DatabaseQueryTool(BaseTool):
    """数据库查询工具"""
    
    name: str = "database_query"
    description: str = "查询数据库获取信息"
    
    def _run(self, query: str) -> str:
        """执行数据库查询"""
        # 实际应该执行 SQL 查询
        return f"查询结果：{query}"


# 使用自定义工具
tools = [
    WeatherTool(),
    DatabaseQueryTool(),
]

5.3 带记忆的 Agent

from langchain.memory import ConversationBufferMemory
from langchain.agents import create_react_agent, AgentExecutor


def build_agent_with_memory():
    """带记忆的 Agent"""
    
    # 记忆
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True,
    )
    
    # Agent
    llm = ChatOpenAI(model="gpt-4")
    
    # 修改 prompt 以包含历史
    prompt = """你是一个有帮助的助手。

历史对话：
{chat_history}

可用工具：
{tools}

{tool_names}

当前任务：{input}

{agent_scratchpad}"""
    
    agent = create_react_agent(llm, tools, prompt)
    
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        memory=memory,
        verbose=True,
    )
    
    return agent_executor


# 使用
agent = build_agent_with_memory()

# 多轮对话
agent.invoke({"input": "我叫张三"})
agent.invoke({"input": "我是做什么的？提示：我是程序员"})
agent.invoke({"input": "你还记得我的名字吗？"})  # Agent 会记得

5.4 LangGraph：更灵活的 Agent 编排

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


class AgentState(TypedDict):
    """Agent 状态"""
    messages: Annotated[list, operator.add]
    next_action: str


def build_langgraph_agent():
    """使用 LangGraph 构建 Agent"""
    
    # 创建状态图
    workflow = StateGraph(AgentState)
    
    # 定义节点
    def router(state: AgentState) -> str:
        """路由决策"""
        last_message = state["messages"][-1]
        if "搜索" in last_message:
            return "search"
        elif "计算" in last_message:
            return "calculate"
        else:
            return "respond"
    
    def search_node(state: AgentState) -> AgentState:
        """搜索节点"""
        result = search_web(state["messages"][-1])
        return {"messages": [result], "next_action": "respond"}
    
    def calculate_node(state: AgentState) -> AgentState:
        """计算节点"""
        result = calculate(state["messages"][-1])
        return {"messages": [result], "next_action": "respond"}
    
    def respond_node(state: AgentState) -> AgentState:
        """响应节点"""
        response = llm.generate(state["messages"])
        return {"messages": [response], "next_action": "end"}
    
    # 添加节点
    workflow.add_node("router", router)
    workflow.add_node("search", search_node)
    workflow.add_node("calculate", calculate_node)
    workflow.add_node("respond", respond_node)
    
    # 添加边
    workflow.add_conditional_edges(
        "router",
        lambda x: x["next_action"],
        {
            "search": "search",
            "calculate": "calculate",
            "respond": "respond",
        }
    )
    workflow.add_edge("search", "respond")
    workflow.add_edge("calculate", "respond")
    workflow.add_edge("respond", END)
    
    # 设置入口
    workflow.set_entry_point("router")
    
    # 编译
    app = workflow.compile()
    
    return app

六、Agent 最佳实践

6.1 工具设计原则

# 好的工具设计
good_tool = Tool(
    name="search_web",
    description="""搜索网页获取信息。
    
    输入: 搜索关键词（字符串）
    输出: 搜索结果摘要
    
    示例输入: "Python 教程"
    示例输出: "找到以下 Python 教程: 1. 官方教程... 2. 菜鸟教程..."
    
    注意: 只返回前 5 条结果的摘要""",
    func=search_web,
)

# 不好的工具设计
bad_tool = Tool(
    name="search",  # 太模糊
    description="搜索东西",  # 描述不清
    func=search_web,
)

6.2 Prompt 工程

AGENT_SYSTEM_PROMPT = """你是一个智能助手，能够使用工具来完成任务。

## 可用工具
{tools}

## 工作原则
1. 仔细分析用户需求，理解真正的意图
2. 制定清晰的解决方案
3. 选择最合适的工具
4. 如果工具返回错误，尝试其他方法
5. 如果无法完成，诚实告知用户

## 输出格式
思考: [分析问题和计划]
行动: [工具名称]
参数: [工具参数，JSON 格式]
---
观察结果会在这里显示
---
思考: [基于观察继续分析]
...

当得到最终答案时:
答案: [最终答案]

## 注意事项
- 不要编造信息
- 如果不确定，先搜索验证
- 复杂任务分步骤执行
- 保持回答简洁准确

现在开始处理用户的请求。
"""

6.3 错误处理和重试

class RobustAgent:
    """带错误处理的 Agent"""
    
    def __init__(self, llm, tools, max_retries=3):
        self.llm = llm
        self.tools = tools
        self.max_retries = max_retries
    
    def execute_tool(self, tool_name: str, tool_input: dict) -> str:
        """执行工具，带重试"""
        
        for attempt in range(self.max_retries):
            try:
                tool = self.tools[tool_name]
                result = tool(**tool_input)
                return result
            
            except Exception as e:
                if attempt < self.max_retries - 1:
                    # 告诉 LLM 出错了，让它尝试修复
                    error_msg = f"工具执行出错: {str(e)}。请尝试其他方法。"
                    return error_msg
                else:
                    return f"工具执行失败: {str(e)}"
    
    def run(self, task: str) -> str:
        """运行 Agent"""
        
        messages = [{"role": "user", "content": task}]
        
        for _ in range(10):  # 最大迭代次数
            response = self.llm.chat(messages)
            
            if self.is_final_answer(response):
                return self.extract_answer(response)
            
            tool_call = self.extract_tool_call(response)
            if tool_call:
                result = self.execute_tool(tool_call["name"], tool_call["input"])
                messages.append({"role": "assistant", "content": response})
                messages.append({"role": "user", "content": f"工具结果: {result}"})
        
        return "达到最大迭代次数"

6.4 评估和监控

import time
from dataclasses import dataclass
from typing import List


@dataclass
class AgentTrace:
    """Agent 执行轨迹"""
    task: str
    steps: List[dict]
    final_answer: str
    total_time: float
    tool_calls: int
    success: bool


class MonitoredAgent:
    """带监控的 Agent"""
    
    def __init__(self, agent):
        self.agent = agent
        self.traces: List[AgentTrace] = []
    
    def run(self, task: str) -> str:
        """运行并记录轨迹"""
        
        start_time = time.time()
        steps = []
        tool_calls = 0
        
        # 包装工具以记录调用
        original_execute = self.agent.execute_tool
        
        def tracked_execute(name, input):
            nonlocal tool_calls
            tool_calls += 1
            result = original_execute(name, input)
            steps.append({
                "type": "tool_call",
                "tool": name,
                "input": input,
                "output": result,
                "timestamp": time.time(),
            })
            return result
        
        self.agent.execute_tool = tracked_execute
        
        try:
            answer = self.agent.run(task)
            success = True
        except Exception as e:
            answer = str(e)
            success = False
        
        # 记录轨迹
        trace = AgentTrace(
            task=task,
            steps=steps,
            final_answer=answer,
            total_time=time.time() - start_time,
            tool_calls=tool_calls,
            success=success,
        )
        self.traces.append(trace)
        
        return answer
    
    def get_metrics(self) -> dict:
        """获取统计指标"""
        if not self.traces:
            return {}
        
        return {
            "total_tasks": len(self.traces),
            "success_rate": sum(t.success for t in self.traces) / len(self.traces),
            "avg_time": sum(t.total_time for t in self.traces) / len(self.traces),
            "avg_tool_calls": sum(t.tool_calls for t in self.traces) / len(self.traces),
        }

七、总结

Agent 核心要点

关键 Takeaway

1. Agent = LLM + 工具 + 规划 + 记忆：不仅能说，还能做
2. ReAct 是基础范式：思考-行动-观察循环
3. 工具设计很重要：清晰的名称、详细的描述、明确的边界
4. 错误处理必不可少：Agent 会出错，要有重试和恢复机制
5. 多 Agent 协作：复杂任务可以分解给多个专业 Agent

6. 选择合适的框架：

   • 简单场景：OpenAI Function Calling
   • 复杂场景：LangChain / LangGraph
   • 多 Agent：AutoGen / CrewAI

下一步学习

• [ ] 多模态 Agent：处理图片、语音
• [ ] 代码执行 Agent：自动写代码并运行
• [ ] 浏览器 Agent：自动操作网页

参考资料

1. ReAct 论文 - Reasoning + Acting
2. AutoGen - 微软多 Agent 框架
3. LangChain Agents - LangChain 文档
4. LangGraph - 复杂 Agent 编排
5. Function Calling - OpenAI 文档