引言

2025年，AI Agent（智能体）已经从概念走向落地。从能够自主浏览网页的WebGPT，到能够编写和执行代码的Devin，AI Agent正在展现出接近人类的自主问题解决能力。

但很多开发者对Agent的理解还停留在”会调用API的聊天机器人”，这显然是不够的。真正的AI Agent具备：

  • 自主规划：能够分解复杂任务
  • 工具使用：会调用各种外部工具
  • 记忆能力：能够学习和积累经验
  • 反思能力：能够从错误中学习

本文将深入探讨AI Agent的核心技术和开发实战，帮助你构建真正智能的Agent系统。

一、AI Agent核心概念

1.1 什么是AI Agent？

传统LLM应用 vs AI Agent：

1
2
3
4
5
传统LLM应用：
用户输入 → LLM处理 → 返回结果

AI Agent：
用户输入 → Agent规划 → 调用工具 → 观察结果 → 决策下一步 → ... → 完成任务

AI Agent的核心特征：

  1. 自主性（Autonomy）：无需人工干预，自主完成任务
  2. 反应性（Reactivity）：能够感知环境变化并做出反应
  3. 主动性（Proactivity）：能够主动规划和执行任务
  4. 社交性（Social Ability）：能够与其他Agent或人类协作

1.2 Agent的核心架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
┌─────────────────────────────────────┐
│          Agent Core                 │
│  ┌──────────────────────────────┐  │
│  │     LLM (大脑)               │  │
│  └──────────────────────────────┘  │
│                                     │
│  ┌──────────┐  ┌──────────────┐   │
│  │  Memory  │  │   Planning    │   │
│  │  (记忆)  │  │   (规划)      │   │
│  └──────────┘  └──────────────┘   │
│                                     │
│  ┌──────────────────────────────┐  │
│  │      Tool Library            │  │
│  │  (工具库)                    │  │
│  │  - Web Search                │  │
│  │  - Code Execution            │  │
│  │  - API Calls                 │  │
│  │  - Database Query            │  │
│  └──────────────────────────────┘  │
└─────────────────────────────────────┘

1.3 主流Agent范式

1. ReAct（Reasoning + Acting）

1
思考(Thought) → 行动(Action) → 观察(Observation) → 思考 → ...

2. Plan-and-Execute

1
规划(Plan) → 执行(Execute) → 验证(Verify) → 调整(Adjust)

3. Reflexion

1
执行 → 失败 → 反思 → 改进策略 → 重新执行

二、ReAct Agent实现详解

2.1 ReAct核心原理

ReAct结合了推理（Reasoning）和行动（Acting），让AI既能思考又能行动。

经典案例：查询某个城市的天气

1
2
3
4
5
6
7
8
9
10
User: 帮我查一下北京明天的天气

Thought: 我需要调用天气API来获取北京明天的天气信息
Action: call_weather_api
Action Input: {"city": "北京", "date": "明天"}
Observation: {"temperature": "15-25°C", "weather": "晴", "wind": "3级"}

Thought: 我已经获取到了天气信息，现在可以回复用户了
Action: Final Answer
Action Input: 北京明天天气晴，温度15-25°C，风力3级

2.2 从零实现ReAct Agent

第一步：定义工具（Tools）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from typing import Dict, Any, Callable
import requests

class Tool:
    """工具基类"""
    def __init__(self, name: str, description: str, func: Callable):
        self.name = name
        self.description = description
        self.func = func
    
    def execute(self, **kwargs) -> Any:
        """执行工具"""
        return self.func(**kwargs)

# 定义具体工具
def search_web(query: str) -> str:
    """网页搜索工具"""
    # 实际应该调用真实的搜索API
    return f"关于'{query}'的搜索结果：[模拟结果]"

def calculate(expression: str) -> float:
    """计算器工具"""
    try:
        # 注意：eval有安全风险，生产环境应使用更安全的方式
        result = eval(expression)
        return result
    except Exception as e:
        return f"计算错误: {str(e)}"

def query_database(sql: str) -> list:
    """数据库查询工具"""
    # 模拟数据库查询
    return [{"id": 1, "name": "示例数据"}]

def get_weather(city: str, date: str = "today") -> Dict[str, Any]:
    """天气查询工具"""
    # 实际应该调用天气API
    return {
        "city": city,
        "date": date,
        "temperature": "20-28°C",
        "weather": "晴",
        "humidity": "60%"
    }

# 创建工具实例
tools = [
    Tool(
        name="search_web",
        description="在互联网上搜索信息。输入：搜索关键词。",
        func=search_web
    ),
    Tool(
        name="calculate",
        description="执行数学计算。输入：数学表达式，如 '2 + 3 * 4'。",
        func=calculate
    ),
    Tool(
        name="query_database",
        description="查询数据库。输入：SQL语句。",
        func=query_database
    ),
    Tool(
        name="get_weather",
        description="查询天气信息。输入：城市名称和日期（可选）。",
        func=get_weather
    )
]

第二步：实现Agent核心逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import json
import re
from typing import List, Optional

class ReActAgent:
    """ReAct Agent实现"""
    
    def __init__(self, llm_client, tools: List[Tool], max_iterations: int = 10):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = max_iterations
        self.history = []
    
    def _build_tools_description(self) -> str:
        """构建工具描述"""
        descriptions = []
        for tool in self.tools.values():
            descriptions.append(f"- {tool.name}: {tool.description}")
        return "\n".join(descriptions)
    
    def _build_prompt(self, user_query: str) -> str:
        """构建ReAct Prompt"""
        tools_desc = self._build_tools_description()
        
        prompt = f"""你是一个智能助手，可以使用以下工具来帮助用户：

{tools_desc}

你需要按照以下格式思考和行动：

Thought: 分析当前情况，思考下一步要做什么
Action: 工具名称
Action Input: 工具输入参数（JSON格式）
Observation: 工具执行结果

重复上述过程直到找到答案，然后：

Thought: 我现在知道了最终答案
Final Answer: 最终答案

用户问题: {user_query}

开始！

"""
        
        # 添加历史记录
        for step in self.history:
            prompt += f"\n{step}"
        
        prompt += "\nThought:"
        return prompt
    
    def _parse_llm_output(self, output: str) -> tuple:
        """解析LLM输出"""
        # 提取Thought
        thought_match = re.search(r'Thought:\s*(.+?)(?=\n|$)', output, re.DOTALL)
        thought = thought_match.group(1).strip() if thought_match else ""
        
        # 检查是否是最终答案
        if "Final Answer:" in output:
            answer_match = re.search(r'Final Answer:\s*(.+?)(?=\n|$)', output, re.DOTALL)
            answer = answer_match.group(1).strip() if answer_match else ""
            return "final_answer", answer, thought
        
        # 提取Action
        action_match = re.search(r'Action:\s*(.+?)(?=\n|$)', output)
        action = action_match.group(1).strip() if action_match else ""
        
        # 提取Action Input
        action_input_match = re.search(r'Action Input:\s*(.+?)(?=\n|$)', output, re.DOTALL)
        action_input = action_input_match.group(1).strip() if action_input_match else ""
        
        return "action", action, action_input, thought
    
    def run(self, user_query: str) -> str:
        """运行Agent"""
        self.history = []
        
        for i in range(self.max_iterations):
            print(f"\n=== Iteration {i + 1} ===")
            
            # 1. 构建prompt
            prompt = self._build_prompt(user_query)
            
            # 2. 调用LLM
            llm_output = self.llm.generate(prompt)
            print(f"LLM Output: {llm_output}")
            
            # 3. 解析输出
            parsed = self._parse_llm_output(llm_output)
            
            if parsed[0] == "final_answer":
                _, answer, thought = parsed
                self.history.append(f"Thought: {thought}")
                self.history.append(f"Final Answer: {answer}")
                return answer
            else:
                _, action, action_input, thought = parsed
                
                # 4. 执行工具
                if action not in self.tools:
                    observation = f"错误：工具 '{action}' 不存在"
                else:
                    try:
                        # 解析JSON输入
                        params = json.loads(action_input) if action_input else {}
                        observation = self.tools[action].execute(**params)
                    except Exception as e:
                        observation = f"工具执行错误: {str(e)}"
                
                print(f"Observation: {observation}")
                
                # 5. 更新历史
                self.history.append(f"Thought: {thought}")
                self.history.append(f"Action: {action}")
                self.history.append(f"Action Input: {action_input}")
                self.history.append(f"Observation: {observation}")
        
        return "抱歉，我无法在限定步骤内完成任务。"

第三步：使用Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 模拟LLM客户端
class MockLLM:
    def generate(self, prompt: str) -> str:
        # 实际应该调用真实的LLM API（OpenAI、Claude等）
        # 这里为了演示，返回模拟输出
        if "天气" in prompt:
            return """Thought: 用户想知道天气，我需要使用get_weather工具
Action: get_weather
Action Input: {"city": "北京", "date": "today"}"""
        return "Final Answer: 这是一个模拟回答"

# 创建并运行Agent
llm = MockLLM()
agent = ReActAgent(llm, tools)

result = agent.run("帮我查一下北京今天的天气")
print(f"\n最终答案: {result}")

2.3 使用LangChain实现ReAct Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain import hub

# 1. 定义工具
def get_current_weather(location: str) -> str:
    """获取指定位置的当前天气"""
    # 实际应该调用天气API
    return f"{location}当前天气：晴，温度25°C"

def search_wikipedia(query: str) -> str:
    """在维基百科搜索"""
    # 实际应该调用维基百科API
    return f"关于{query}的维基百科信息..."

tools = [
    Tool(
        name="GetWeather",
        func=get_current_weather,
        description="获取指定城市的当前天气。输入应该是城市名称。"
    ),
    Tool(
        name="SearchWikipedia",
        func=search_wikipedia,
        description="在维基百科搜索信息。输入应该是搜索关键词。"
    )
]

# 2. 创建LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

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

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

# 5. 创建Agent Executor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

# 6. 运行Agent
result = agent_executor.invoke({
    "input": "告诉我北京的天气，然后查一下北京的历史"
})

print(result["output"])

三、高级Agent模式

3.1 Plan-and-Execute Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from typing import List, Dict

class PlanAndExecuteAgent:
    """计划-执行型Agent"""
    
    def __init__(self, llm_client, tools: List[Tool]):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
    
    def plan(self, user_query: str) -> List[Dict]:
        """生成执行计划"""
        prompt = f"""
将以下任务分解为详细的执行步骤：

任务: {user_query}

可用工具:
{self._build_tools_description()}

请返回JSON格式的执行计划：
[
    {{"step": 1, "action": "工具名", "params": {{}}, "description": "描述"}},
    ...
]
"""
        plan_json = self.llm.generate(prompt)
        return json.loads(plan_json)
    
    def execute(self, plan: List[Dict]) -> str:
        """执行计划"""
        results = []
        
        for step in plan:
            print(f"\n执行步骤 {step['step']}: {step['description']}")
            
            tool_name = step['action']
            params = step['params']
            
            if tool_name in self.tools:
                result = self.tools[tool_name].execute(**params)
                results.append({
                    "step": step['step'],
                    "result": result
                })
                print(f"结果: {result}")
            else:
                results.append({
                    "step": step['step'],
                    "error": f"工具{tool_name}不存在"
                })
        
        # 综合所有结果生成最终答案
        summary_prompt = f"""
基于以下执行结果，生成最终答案：

{json.dumps(results, ensure_ascii=False, indent=2)}
"""
        final_answer = self.llm.generate(summary_prompt)
        return final_answer
    
    def run(self, user_query: str) -> str:
        """运行Agent"""
        # 1. 规划
        plan = self.plan(user_query)
        print(f"执行计划: {json.dumps(plan, ensure_ascii=False, indent=2)}")
        
        # 2. 执行
        result = self.execute(plan)
        return result

3.2 Reflexion Agent（反思型）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class ReflexionAgent:
    """具备反思能力的Agent"""
    
    def __init__(self, llm_client, tools: List[Tool], max_retries: int = 3):
        self.llm = llm_client
        self.tools = tools
        self.max_retries = max_retries
        self.reflection_history = []
    
    def execute_with_reflection(self, user_query: str) -> str:
        """带反思的执行"""
        for attempt in range(self.max_retries):
            print(f"\n=== 尝试 {attempt + 1} ===")
            
            # 执行任务
            result, success = self._execute_task(user_query)
            
            if success:
                return result
            
            # 失败则进行反思
            reflection = self._reflect_on_failure(user_query, result)
            self.reflection_history.append(reflection)
            print(f"反思: {reflection}")
        
        return "经过多次尝试和反思，仍未能完成任务。"
    
    def _execute_task(self, user_query: str) -> tuple:
        """执行任务"""
        # 加入之前的反思经验
        context = ""
        if self.reflection_history:
            context = f"之前的经验教训：\n" + "\n".join(self.reflection_history)
        
        prompt = f"""
{context}

现在请完成任务: {user_query}
"""
        result = self.llm.generate(prompt)
        
        # 验证结果（这里简化处理）
        success = self._validate_result(result)
        return result, success
    
    def _reflect_on_failure(self, query: str, failed_result: str) -> str:
        """对失败进行反思"""
        prompt = f"""
任务: {query}
执行结果: {failed_result}

这次执行失败了。请分析失败原因，并给出改进建议：
1. 哪里出错了？
2. 应该怎么改进？
3. 下次应该避免什么？
"""
        reflection = self.llm.generate(prompt)
        return reflection
    
    def _validate_result(self, result: str) -> bool:
        """验证结果是否正确"""
        # 实际应该有更复杂的验证逻辑
        return "错误" not in result and len(result) > 10

四、多智能体系统（Multi-Agent System）

4.1 多Agent协作架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
from enum import Enum
from dataclasses import dataclass
from typing import Optional

class AgentRole(Enum):
    """Agent角色"""
    PLANNER = "规划者"
    RESEARCHER = "研究员"
    CODER = "程序员"
    REVIEWER = "审查员"
    EXECUTOR = "执行者"

@dataclass
class Message:
    """Agent间通信消息"""
    from_agent: str
    to_agent: str
    content: str
    task_id: str

class BaseAgent:
    """Agent基类"""
    
    def __init__(self, name: str, role: AgentRole, llm_client, tools: List[Tool] = None):
        self.name = name
        self.role = role
        self.llm = llm_client
        self.tools = tools or []
        self.message_queue = []
    
    def receive_message(self, message: Message):
        """接收消息"""
        self.message_queue.append(message)
    
    def process_message(self, message: Message) -> Optional[Message]:
        """处理消息"""
        raise NotImplementedError
    
    def send_message(self, to_agent: str, content: str, task_id: str) -> Message:
        """发送消息"""
        return Message(
            from_agent=self.name,
            to_agent=to_agent,
            content=content,
            task_id=task_id
        )

class PlannerAgent(BaseAgent):
    """规划Agent"""
    
    def process_message(self, message: Message) -> Optional[Message]:
        """制定任务计划"""
        prompt = f"""
作为项目规划者，请为以下任务制定详细计划：

任务: {message.content}

请分解为子任务，并指定负责的Agent（可选：Researcher/Coder/Reviewer/Executor）
"""
        plan = self.llm.generate(prompt)
        
        # 将计划发送给执行者
        return self.send_message(
            to_agent="Executor",
            content=plan,
            task_id=message.task_id
        )

class ResearcherAgent(BaseAgent):
    """研究Agent"""
    
    def process_message(self, message: Message) -> Optional[Message]:
        """进行信息研究"""
        prompt = f"""
作为研究员，请收集以下主题的相关信息：

主题: {message.content}

提供详细的背景资料和技术建议。
"""
        research = self.llm.generate(prompt)
        
        return self.send_message(
            to_agent=message.from_agent,
            content=research,
            task_id=message.task_id
        )

class CoderAgent(BaseAgent):
    """编码Agent"""
    
    def process_message(self, message: Message) -> Optional[Message]:
        """编写代码"""
        prompt = f"""
作为程序员，请实现以下功能：

需求: {message.content}

提供完整的代码实现和必要的注释。
"""
        code = self.llm.generate(prompt)
        
        # 将代码发送给审查员
        return self.send_message(
            to_agent="Reviewer",
            content=code,
            task_id=message.task_id
        )

class ReviewerAgent(BaseAgent):
    """审查Agent"""
    
    def process_message(self, message: Message) -> Optional[Message]:
        """审查代码或方案"""
        prompt = f"""
作为审查员，请仔细审查以下内容：

{message.content}

检查：
1. 是否有错误或bug
2. 是否符合最佳实践
3. 是否有改进空间

提供详细的审查报告。
"""
        review = self.llm.generate(prompt)
        
        return self.send_message(
            to_agent=message.from_agent,
            content=review,
            task_id=message.task_id
        )

class MultiAgentSystem:
    """多Agent系统orchestrator"""
    
    def __init__(self, agents: List[BaseAgent]):
        self.agents = {agent.name: agent for agent in agents}
        self.task_results = {}
    
    def execute_task(self, task: str, task_id: str) -> str:
        """执行任务"""
        # 1. 从Planner开始
        initial_message = Message(
            from_agent="User",
            to_agent="Planner",
            content=task,
            task_id=task_id
        )
        
        current_message = initial_message
        max_iterations = 20
        iteration = 0
        
        while iteration < max_iterations:
            iteration += 1
            print(f"\n=== 回合 {iteration} ===")
            print(f"{current_message.from_agent} -> {current_message.to_agent}: {current_message.content[:100]}...")
            
            # 找到目标Agent
            target_agent = self.agents.get(current_message.to_agent)
            if not target_agent:
                break
            
            # 处理消息
            target_agent.receive_message(current_message)
            next_message = target_agent.process_message(current_message)
            
            if not next_message:
                # 任务完成
                self.task_results[task_id] = current_message.content
                return current_message.content
            
            current_message = next_message
        
        return "任务执行超时"

4.2 使用AutoGen实现多Agent系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import autogen

# 配置LLM
config_list = [{
    "model": "gpt-4",
    "api_key": "your-api-key"
}]

llm_config = {
    "config_list": config_list,
    "temperature": 0.7
}

# 创建用户代理
user_proxy = autogen.UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

# 创建规划Agent
planner = autogen.AssistantAgent(
    name="Planner",
    llm_config=llm_config,
    system_message="""你是项目规划专家。负责：
1. 分析用户需求
2. 制定详细的执行计划
3. 分配任务给其他Agent
"""
)

# 创建编码Agent
coder = autogen.AssistantAgent(
    name="Coder",
    llm_config=llm_config,
    system_message="""你是资深程序员。负责：
1. 根据规划编写高质量代码
2. 遵循最佳实践
3. 添加必要的注释和文档
"""
)

# 创建审查Agent
reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config=llm_config,
    system_message="""你是代码审查专家。负责：
1. 审查代码质量
2. 发现潜在问题
3. 提出改进建议
"""
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, planner, coder, reviewer],
    messages=[],
    max_round=20
)

manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# 执行任务
user_proxy.initiate_chat(
    manager,
    message="开发一个Python的Todo List命令行应用，支持增删改查和数据持久化"
)

五、Agent工具系统设计

5.1 工具注册与发现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from typing import Callable, Any, Dict
from functools import wraps
import inspect

class ToolRegistry:
    """工具注册中心"""
    
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
    
    def register(self, 
                 name: str = None,
                 description: str = None,
                 parameters_schema: Dict = None):
        """工具注册装饰器"""
        def decorator(func: Callable) -> Callable:
            tool_name = name or func.__name__
            tool_description = description or func.__doc__ or ""
            
            # 自动生成参数schema
            if parameters_schema is None:
                sig = inspect.signature(func)
                auto_schema = {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
                
                for param_name, param in sig.parameters.items():
                    param_type = "string"  # 简化处理
                    auto_schema["properties"][param_name] = {
                        "type": param_type,
                        "description": f"Parameter {param_name}"
                    }
                    if param.default == inspect.Parameter.empty:
                        auto_schema["required"].append(param_name)
                
                schema = auto_schema
            else:
                schema = parameters_schema
            
            tool = Tool(
                name=tool_name,
                description=tool_description,
                func=func,
                parameters_schema=schema
            )
            
            self.tools[tool_name] = tool
            
            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            
            return wrapper
        
        return decorator
    
    def get_tool(self, name: str) -> Optional[Tool]:
        """获取工具"""
        return self.tools.get(name)
    
    def list_tools(self) -> List[Tool]:
        """列出所有工具"""
        return list(self.tools.values())
    
    def to_openai_functions(self) -> List[Dict]:
        """转换为OpenAI Function Calling格式"""
        functions = []
        for tool in self.tools.values():
            functions.append({
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters_schema
            })
        return functions

# 使用示例
registry = ToolRegistry()

@registry.register(
    name="search_web",
    description="在互联网上搜索信息",
    parameters_schema={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词"
            },
            "max_results": {
                "type": "integer",
                "description": "最大结果数",
                "default": 10
            }
        },
        "required": ["query"]
    }
)
def search_web(query: str, max_results: int = 10) -> List[Dict]:
    """执行网页搜索"""
    # 实际实现
    return []

@registry.register(description="获取当前天气")
def get_weather(city: str) -> Dict[str, Any]:
    """获取指定城市的天气"""
    return {"temperature": "25°C", "weather": "晴"}

# 列出所有工具
print("Available tools:", [tool.name for tool in registry.list_tools()])

# 转换为OpenAI格式
print("OpenAI functions:", registry.to_openai_functions())

5.2 使用Function Calling

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import openai
import json

class FunctionCallingAgent:
    """基于OpenAI Function Calling的Agent"""
    
    def __init__(self, api_key: str, tool_registry: ToolRegistry):
        self.client = openai.OpenAI(api_key=api_key)
        self.registry = tool_registry
        self.conversation_history = []
    
    def run(self, user_message: str) -> str:
        """运行Agent"""
        # 添加用户消息
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        max_iterations = 10
        
        for i in range(max_iterations):
            print(f"\n=== 迭代 {i + 1} ===")
            
            # 调用OpenAI
            response = self.client.chat.completions.create(
                model="gpt-4-turbo-preview",
                messages=self.conversation_history,
                functions=self.registry.to_openai_functions(),
                function_call="auto"
            )
            
            message = response.choices[0].message
            
            # 检查是否需要调用函数
            if message.function_call:
                function_name = message.function_call.name
                function_args = json.loads(message.function_call.arguments)
                
                print(f"调用工具: {function_name}")
                print(f"参数: {function_args}")
                
                # 执行工具
                tool = self.registry.get_tool(function_name)
                if tool:
                    result = tool.execute(**function_args)
                    print(f"结果: {result}")
                    
                    # 将结果添加到对话历史
                    self.conversation_history.append({
                        "role": "function",
                        "name": function_name,
                        "content": json.dumps(result, ensure_ascii=False)
                    })
                else:
                    self.conversation_history.append({
                        "role": "function",
                        "name": function_name,
                        "content": json.dumps({"error": "Tool not found"})
                    })
            else:
                # 没有函数调用，返回最终答案
                final_answer = message.content
                self.conversation_history.append({
                    "role": "assistant",
                    "content": final_answer
                })
                return final_answer
        
        return "达到最大迭代次数"

# 使用
agent = FunctionCallingAgent(api_key="your-key", tool_registry=registry)
result = agent.run("帮我查一下北京的天气，然后在网上搜索北京最近的新闻")
print(f"\n最终答案: {result}")

六、Agent记忆系统

6.1 短期记忆（对话历史）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from collections import deque

class ShortTermMemory:
    """短期记忆（对话窗口）"""
    
    def __init__(self, max_messages: int = 10):
        self.messages = deque(maxlen=max_messages)
    
    def add(self, role: str, content: str):
        """添加消息"""
        self.messages.append({"role": role, "content": content})
    
    def get_context(self) -> List[Dict]:
        """获取上下文"""
        return list(self.messages)
    
    def clear(self):
        """清空记忆"""
        self.messages.clear()

6.2 长期记忆（向量数据库）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from typing import List
import numpy as np
from sentence_transformers import SentenceTransformer

class LongTermMemory:
    """长期记忆（基于向量检索）"""
    
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(embedding_model)
        self.memories = []
        self.embeddings = []
    
    def store(self, content: str, metadata: Dict = None):
        """存储记忆"""
        embedding = self.model.encode(content)
        self.memories.append({
            "content": content,
            "metadata": metadata or {},
            "timestamp": time.time()
        })
        self.embeddings.append(embedding)
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
        """检索相关记忆"""
        if not self.memories:
            return []
        
        query_embedding = self.model.encode(query)
        
        # 计算相似度
        similarities = []
        for emb in self.embeddings:
            sim = np.dot(query_embedding, emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(emb)
            )
            similarities.append(sim)
        
        # 获取top_k
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        
        return [self.memories[i] for i in top_indices]

6.3 混合记忆Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class MemoryEnhancedAgent:
    """具备记忆能力的Agent"""
    
    def __init__(self, llm_client, tools: List[Tool]):
        self.llm = llm_client
        self.tools = tools
        self.short_term = ShortTermMemory(max_messages=20)
        self.long_term = LongTermMemory()
    
    def run(self, user_query: str) -> str:
        """运行Agent"""
        # 1. 从长期记忆检索相关经验
        relevant_memories = self.long_term.retrieve(user_query, top_k=3)
        
        # 2. 构建带记忆的prompt
        memory_context = ""
        if relevant_memories:
            memory_context = "相关经验：\n"
            for mem in relevant_memories:
                memory_context += f"- {mem['content']}\n"
        
        # 3. 获取短期记忆（对话历史）
        conversation_context = self.short_term.get_context()
        
        # 4. 执行任务
        prompt = f"""
{memory_context}

对话历史：
{json.dumps(conversation_context, ensure_ascii=False, indent=2)}

用户新问题: {user_query}

请基于以上信息回答。
"""
        
        result = self.llm.generate(prompt)
        
        # 5. 更新记忆
        self.short_term.add("user", user_query)
        self.short_term.add("assistant", result)
        
        # 如果是重要信息，存入长期记忆
        if self._is_important(user_query, result):
            self.long_term.store(
                f"Q: {user_query}\nA: {result}",
                metadata={"type": "qa_pair"}
            )
        
        return result
    
    def _is_important(self, query: str, result: str) -> bool:
        """判断是否值得存入长期记忆"""
        # 简化逻辑：如果回答较长，认为重要
        return len(result) > 100

七、Agent评估与优化

7.1 评估指标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from dataclasses import dataclass
from typing import List

@dataclass
class AgentMetrics:
    """Agent评估指标"""
    success_rate: float  # 成功率
    avg_steps: float     # 平均步数
    avg_time: float      # 平均耗时
    avg_tokens: int      # 平均token消耗
    avg_cost: float      # 平均成本

class AgentEvaluator:
    """Agent评估器"""
    
    def __init__(self, test_cases: List[Dict]):
        self.test_cases = test_cases
        self.results = []
    
    def evaluate(self, agent) -> AgentMetrics:
        """评估Agent"""
        total_success = 0
        total_steps = 0
        total_time = 0
        total_tokens = 0
        total_cost = 0
        
        for test_case in self.test_cases:
            start_time = time.time()
            
            try:
                result = agent.run(test_case['input'])
                end_time = time.time()
                
                # 判断是否成功
                success = self._check_success(
                    result,
                    test_case['expected']
                )
                
                if success:
                    total_success += 1
                
                # 记录指标
                total_steps += agent.get_step_count()
                total_time += (end_time - start_time)
                total_tokens += agent.get_token_usage()
                total_cost += agent.get_cost()
                
                self.results.append({
                    "input": test_case['input'],
                    "output": result,
                    "success": success,
                    "time": end_time - start_time
                })
            
            except Exception as e:
                self.results.append({
                    "input": test_case['input'],
                    "error": str(e),
                    "success": False
                })
        
        n = len(self.test_cases)
        return AgentMetrics(
            success_rate=total_success / n,
            avg_steps=total_steps / n,
            avg_time=total_time / n,
            avg_tokens=total_tokens / n,
            avg_cost=total_cost / n
        )
    
    def _check_success(self, result: str, expected: Any) -> bool:
        """检查结果是否符合预期"""
        # 实际应该有更复杂的验证逻辑
        if isinstance(expected, str):
            return expected.lower() in result.lower()
        return True

八、最佳实践与注意事项

8.1 Agent开发最佳实践

  1. 从简单开始，逐步迭代

    1
    2
    3
    4
    # 第一版：基础ReAct Agent
    # 第二版：添加记忆
    # 第三版：添加反思
    # 第四版：多Agent协作

  2. 合理设置循环次数限制

    1
    MAX_ITERATIONS = 15  # 防止无限循环

  3. 完善的错误处理

    1
    2
    3
    4
    5
    try:
        result = tool.execute(**params)
    except Exception as e:
        logger.error(f"Tool execution failed: {e}")
        result = f"Error: {str(e)}"

  4. 日志和可观测性

    1
    2
    3
    4
    5
    import logging

    logger = logging.getLogger("agent")
    logger.info(f"Agent step {i}: {action}")
    logger.debug(f"Observation: {observation}")

8.2 常见陷阱

  1. 无限循环：Agent陷入重复的思考-行动循环
  2. Prompt污染：历史记录过长导致LLM混乱
  3. 工具滥用：频繁调用高成本工具
  4. 上下文丢失：多轮对话后忘记初始目标

8.3 成本优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 1. 使用更便宜的模型处理简单任务
def select_model(task_complexity: str) -> str:
    if task_complexity == "simple":
        return "gpt-3.5-turbo"  # 便宜
    elif task_complexity == "medium":
        return "gpt-4"
    else:
        return "gpt-4-turbo"    # 昂贵但能力强

# 2. 缓存重复查询
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_llm_call(prompt: str) -> str:
    return llm.generate(prompt)

# 3. 批量处理
def batch_process(queries: List[str]) -> List[str]:
    # 一次API调用处理多个查询
    pass

九、总结与展望

9.1 核心要点回顾

  1. AI Agent ≠ 简单的LLM调用：需要规划、工具使用、记忆等能力
  2. 选择合适的Agent范式：ReAct适合通用任务，Plan-and-Execute适合复杂任务
  3. 工具系统是关键：丰富、可靠的工具决定了Agent的能力上限
  4. 记忆增强Agent能力：短期记忆维持上下文，长期记忆积累经验
  5. 多Agent协作解决复杂问题：分工明确，各司其职

9.2 未来发展方向

  1. 自主学习：Agent能从经验中学习，持续改进
  2. 多模态Agent：处理文本、图像、音频、视频
  3. 具身Agent：控制机器人等物理实体
  4. 通用AI Agent：一个Agent解决所有问题

9.3 入门建议

  1. 从ReAct开始：最容易理解和实现
  2. 使用现成框架：LangChain、AutoGen、CrewAI
  3. 构建自己的工具库：针对特定领域
  4. 参与开源项目：学习优秀实践

参考资源

AI Agent不是科幻，而是正在改变世界的现实技术。掌握Agent开发，就是掌握AI应用的未来。