深入理解 AI Agent 架构：从单 Agent 到多 Agent 协作

1
         ┌─────────────────────────────────────┐
2
         │           AI Agent                  │
3
         │                                      │
4
         │   ┌────────────┐  ┌────────────┐   │
5
         │   │ Perception │  │  Memory    │   │
6
         │   └─────┬──────┘  └─────▲──────┘   │
7
         │         │               │          │
8
         │         ▼               │          │
9
         │   ┌─────────────────────┴────────┐ │
10
         │   │        Reasoning             │ │
11
         │   │   (LLM + Planning)          │ │
12
         │   └──────────────┬───────────────┘ │
13
         │                  │                 │
14
         │                  ▼                 │
15
         │         ┌──────────────┐          │
16
         │         │    Action    │          │
17
         │         └──────┬───────┘          │
18
         └────────────────┼──────────────────┘
19
                          ▼
20
                    [Environment]

1
class Perception:
2
    """Agent 感知模块"""
3
    def __init__(self):
4
        self.input_processors = []
5

6
    def add_processor(self, processor):
7
        """注册输入处理器"""
8
        self.input_processors.append(processor)
9

10
    def perceive(self, raw_input):
11
        """处理输入"""
12
        processed = raw_input
13

14
        for processor in self.input_processors:
15
            processed = processor(processed)
16

17
        return processed
18

19
# 示例: 多模态感知
20
perception = Perception()
21

22
# 文本输入
23
perception.add_processor(lambda x: x.strip())
24

25
# 包含图像
26
def image_processor(input):
27
    if "<image>" in input:
28
        # 调用视觉模型描述图像
29
        return describe_image(input)
30
    return input
31
perception.add_processor(image_processor)

1
class Memory:
2
    """Agent 记忆系统"""
3
    def __init__(self):
4
        self.short_term = []      # 短期记忆 (当前任务)
5
        self.long_term = {}        # 长期记忆 (向量化)
6
        self.episodic = []         # 情节记忆 (历史行动)
7
        self.semantic = {}         # 语义记忆 (事实知识)
8

9
    def store(self, content, memory_type='short_term'):
10
        """存储记忆"""
11
        if memory_type == 'short_term':
12
            self.short_term.append(content)
13
        elif memory_type == 'long_term':
14
            vector = self.embed(content)
15
            self.long_term[vector] = content
16
        # ...
17

18
    def retrieve(self, query, top_k=5):
19
        """检索相关记忆"""
20
        # 使用 embedding 相似度
21
        query_vector = self.embed(query)
22
        relevant = self.similarity_search(query_vector, top_k)
23
        return relevant

1
class Reasoning:
2
    """Agent 推理模块"""
3
    def __init__(self, llm):
4
        self.llm = llm
5
        self.planner = Planner(llm)
6
        self.reflector = Reflector(llm)
7

8
    def reason(self, state, goal):
9
        """推理下一步行动"""
10
        # 1. 规划：分解目标
11
        plan = self.planner.create_plan(state, goal)
12

13
        # 2. 执行当前步骤
14
        step = plan.current_step()
15

16
        # 3. 反思：如果失败，调整
17
        if self._is_stuck(state):
18
            plan = self.reflector.reflect(state, plan)
19
            step = plan.current_step()
20

21
        return step
22

23
class Planner:
24
    """规划器"""
25
    def create_plan(self, state, goal):
26
        prompt = f"""
27
        目标: {goal}
28
        当前状态: {state}
29

30
        请创建一个详细的执行计划：
31
        1. [步骤1]
32
        2. [步骤2]
33
        ...
34
        """
35
        plan_text = self.llm.generate(prompt)
36
        return self._parse_plan(plan_text)

1
class Action:
2
    """Agent 行动模块"""
3
    def __init__(self, tools):
4
        self.tools = tools    # 可用工具集合
5

6
    def execute(self, action):
7
        """执行行动"""
8
        tool_name = action.tool
9
        args = action.arguments
10

11
        tool = self.tools.get(tool_name)
12
        if not tool:
13
            raise ToolNotFound(tool_name)
14

15
        result = tool.execute(**args)
16
        return result
17

18
class Tool:
19
    """工具"""
20
    def __init__(self, name, func, schema):
21
        self.name = name
22
        self.func = func
23
        self.schema = schema
24

25
    def execute(self, **kwargs):
26
        try:
27
            result = self.func(**kwargs)
28
            return ActionResult(success=True, data=result)
29
        except Exception as e:
30
            return ActionResult(success=False, error=str(e))

1
class ReActAgent:
2
    """ReAct Agent"""
3
    def __init__(self, llm, tools, max_steps=10):
4
        self.llm = llm
5
        self.tools = tools
6
        self.max_steps = max_steps
7

8
    def run(self, question):
9
        history = []
10

11
        for step in range(self.max_steps):
12
            # 1. Thought + Action
13
            prompt = self._build_prompt(question, history)
14
            response = self.llm.generate(prompt)
15

16
            thought, action = self._parse_response(response)
17

18
            # 2. 完成检测
19
            if self._is_final_answer(response):
20
                return self._extract_answer(response)
21

22
            # 3. 执行 Action
23
            observation = self._execute_tool(action)
24

25
            # 4. 更新历史
26
            history.append({
27
                'thought': thought,
28
                'action': action,
29
                'observation': observation
30
            })
31

32
        return "达到最大步数"
33

34
# ReAct 循环示意:
35
# Thought 1: 需要查询北京天气
36
# Action 1: get_weather(city="北京")
37
# Observation 1: 25°C, 晴
38
#
39
# Thought 2: 现在查空气质量
40
# Action 2: get_air_quality(city="北京")
41
# Observation 2: AQI 85
42
#
43
# Thought 3: 可以给出建议
44
# Final Answer: 北京今天 25°C, 晴朗, 空气良好...

1
class PlanExecuteAgent:
2
    """Plan-and-Execute Agent"""
3

4
    def run(self, question):
5
        # 1. 规划阶段
6
        plan = self.create_plan(question)
7

8
        # 2. 执行阶段
9
        results = []
10
        for step in plan:
11
            try:
12
                result = self.execute_step(step)
13
                results.append(result)
14
            except Exception as e:
15
                # 失败：重规划
16
                plan = self.replan(question, plan, results, e)
17
                return self.run(question)  # 重试
18

19
        return self.synthesize(results)
20

21
    def create_plan(self, question):
22
        """创建完整计划"""
23
        prompt = f"""
24
        任务: {question}
25

26
        请创建一个详细的执行计划。每个步骤应该明确指定:
27
        - 工具
28
        - 参数
29
        - 预期输出
30

31
        计划:
32
        """
33
        plan_text = self.llm.generate(prompt)
34
        return self._parse_plan(plan_text)
35

36
    def replan(self, question, plan, results, error):
37
        """失败后重新规划"""
38
        prompt = f"""
39
        原任务: {question}
40
        原计划: {plan}
41
        已执行: {results}
42
        错误: {error}
43

44
        请重新规划。跳过失败步骤或找到替代方案。
45

46
        新计划:
47
        """
48
        new_plan_text = self.llm.generate(prompt)
49
        return self._parse_plan(new_plan_text)

1
class ReflexionAgent(ReActAgent):
2
    """带反思的 Agent"""
3

4
    def __init__(self, llm, tools, max_reflections=3):
5
        super().__init__(llm, tools)
6
        self.max_reflections = max_reflections
7
        self.reflections = []
8

9
    def run(self, question):
10
        for attempt in range(self.max_reflections + 1):
11
            # 执行 ReAct
12
            result = super().run(question)
13

14
            # 评估
15
            if self.is_successful(result):
16
                return result
17

18
            if attempt < self.max_reflections:
19
                # 反思
20
                reflection = self.reflect(question, result)
21
                self.reflections.append(reflection)
22

23
        return result
24

25
    def reflect(self, question, result):
26
        """反思失败原因"""
27
        prompt = f"""
28
        问题: {question}
29
        尝试结果: {result}
30

31
        反思以下:
32
        1. 为什么失败？
33
        2. 下次可以采取什么不同的策略？
34
        3. 你学到了什么？
35

36
        反思:
37
        """
38
        return self.llm.generate(prompt)

1
class ReWOOAgent:
2
    """ReWOO Agent - 减少观察次数"""
3

4
    def run(self, question):
5
        # 1. 一次性生成所有步骤
6
        plan = self.generate_full_plan(question)
7

8
        # 2. 执行所有工具调用（无需中间反馈）
9
        evidence = {}
10
        for step in plan:
11
            result = self.execute_tool(step)
12
            evidence[step.id] = result
13

14
        # 3. 综合所有证据
15
        return self.synthesize(question, plan, evidence)
16

17
    def generate_full_plan(self, question):
18
        """生成完整计划，包括变量替换"""
19
        # 计划结构:
20
        # Step 1: Search("Beijing weather") -> #E1
21
        # Step 2: Search("Shanghai weather") -> #E2
22
        # Step 3: LLM(#E1, #E2, "compare weathers") -> #E3
23
        return self.llm.generate(prompt_for_plan(question))

1
class ToTAgent:
2
    """Tree of Thoughts Agent"""
3

4
    def __init__(self, llm, branching_factor=3, max_depth=5):
5
        self.llm = llm
6
        self.branching_factor = branching_factor
7
        self.max_depth = max_depth
8

9
    def run(self, question):
10
        root = ThoughtNode(question)
11
        tree = ThoughtTree(root)
12

13
        for depth in range(self.max_depth):
14
            # 1. 扩展当前层
15
            current_nodes = tree.get_nodes_at_depth(depth)
16
            for node in current_nodes:
17
                # 生成多个候选
18
                thoughts = self.generate_thoughts(node.state, n=self.branching_factor)
19
                for thought in thoughts:
20
                    child = ThoughtNode(thought, parent=node)
21

22
                    # 评估
23
                    child.score = self.evaluate(thought)
24

25
                    tree.add(child)
26

27
            # 2. 剪枝
28
            tree.prune(keep_top_n=self.branching_factor)
29

30
        return tree.get_best_path()
31

32
    def generate_thoughts(self, state, n):
33
        """生成 n 个下一步思考"""
34
        prompt = f"当前: {state}\n生成 {n} 个可能的下一步思考："
35
        response = self.llm.generate(prompt)
36
        return self._parse_thoughts(response, n)
37

38
    def evaluate(self, thought):
39
        """评估思考的质量"""
40
        prompt = f"评估以下思考（1-10）：{thought}"
41
        return int(self.llm.generate(prompt))

1
class TaskDecomposer:
2
    """任务分解器"""
3

4
    def decompose(self, task):
5
        """将复杂任务分解为子任务"""
6
        prompt = f"""
7
        任务: {task}
8

9
        请将这个任务分解为多个子任务。每个子任务应该：
10
        1. 具体明确
11
        2. 可独立执行
12
        3. 有明确的成功标准
13

14
        子任务列表（JSON 数组）:
15
        """
16
        response = self.llm.generate(prompt)
17
        return self._parse_tasks(response)
18

19
# 示例分解:
20
# 原始任务: "研究某个 AI 主题"
21
# 分解为:
22
# 1. 搜索 arXiv 相关论文
23
# 2. 阅读最重要的 3 篇
24
# 3. 对比方法
25
# 4. 总结发现

1
from dataclasses import dataclass
2
from typing import List, Optional
3

4
@dataclass
5
class PlanStep:
6
    """计划步骤"""
7
    id: int
8
    description: str
9
    tool: Optional[str] = None    # 可选的工具
10
    args: dict = None             # 工具参数
11
    depends_on: List[int] = None  # 依赖的前置步骤
12
    status: str = "pending"       # pending/running/done/failed
13
    result: any = None
14

15
@dataclass
16
class Plan:
17
    """完整计划"""
18
    goal: str
19
    steps: List[PlanStep]
20

21
    def ready_steps(self):
22
        """获取所有可执行的步骤（依赖已满足）"""
23
        done_ids = {s.id for s in self.steps if s.status == 'done'}
24
        return [
25
            s for s in self.steps
26
            if s.status == 'pending' and
27
            all(dep in done_ids for dep in (s.depends_on or []))
28
        ]
29

30
    def is_complete(self):
31
        return all(s.status == 'done' for s in self.steps)
32

33
    def progress(self):
34
        done = sum(1 for s in self.steps if s.status == 'done')
35
        return done / len(self.steps)

1
class DynamicPlanner:
2
    """动态规划器"""
3

4
    def run_with_replanning(self, plan, executor):
5
        """带重规划的执行"""
6
        while not plan.is_complete():
7
            ready = plan.ready_steps()
8

9
            # 并行执行所有就绪步骤
10
            futures = []
11
            for step in ready:
12
                step.status = 'running'
13
                futures.append(executor.execute_async(step))
14

15
            # 等待结果
16
            for step, future in zip(ready, futures):
17
                try:
18
                    result = future.result()
19
                    step.status = 'done'
20
                    step.result = result
21
                except Exception as e:
22
                    step.status = 'failed'
23
                    step.result = str(e)
24

25
            # 检查是否需要重规划
26
            failed = [s for s in plan.steps if s.status == 'failed']
27
            if failed:
28
                plan = self.replan(plan, failed)
29

30
        return plan

1
class HierarchicalPlanner:
2
    """分层规划器"""
3

4
    def plan(self, goal):
5
        """分层规划"""
6
        # Level 1: 战略规划
7
        strategy = self.strategic_plan(goal)
8

9
        # Level 2: 战术规划
10
        tactics = [self.tactical_plan(subgoal) for subgoal in strategy]
11

12
        # Level 3: 操作规划
13
        operations = []
14
        for tactic in tactics:
15
            ops = self.operational_plan(tactic)
16
            operations.extend(ops)
17

18
        return Plan(strategy, tactics, operations)
19

20
    def strategic_plan(self, goal):
21
        """高层次规划"""
22
        prompt = f"目标: {goal}\n拆分为 3-5 个主要阶段："
23
        return self.llm.generate(prompt)
24

25
    def tactical_plan(self, subgoal):
26
        """子目标规划"""
27
        prompt = f"子目标: {subgoal}\n具体行动步骤："
28
        return self.llm.generate(prompt)
29

30
    def operational_plan(self, action):
31
        """操作级规划"""
32
        # 确定具体的工具调用
33
        # ...

1
┌─────────────────────────────────────────────────────┐
2
│                  Agent Memory                        │
3
├─────────────────────────────────────────────────────┤
4
│                                                      │
5
│  ┌──────────────────────────────────────────────┐    │
6
│  │         Short-Term Memory                    │    │
7
│  │  - 当前对话上下文                            │    │
8
│  │  - 最近 N 轮对话                            │    │
9
│  │  - 滑动窗口                                 │    │
10
│  └──────────────────────────────────────────────┘    │
11
│                                                      │
12
│  ┌──────────────────────────────────────────────┐    │
13
│  │         Long-Term Memory (向量存储)          │    │
14
│  │  - 历史对话摘要                              │    │
15
│  │  - 用户偏好                                  │    │
16
│  │  - 事实知识                                  │    │
17
│  └──────────────────────────────────────────────┘    │
18
│                                                      │
19
│  ┌──────────────────────────────────────────────┐    │
20
│  │         Episodic Memory                      │    │
21
│  │  - 具体的过去事件                            │    │
22
│  │  - "上次我尝试 X 然后..."                     │    │
23
│  │  - 成功/失败模式                             │    │
24
│  └──────────────────────────────────────────────┘    │
25
│                                                      │
26
│  ┌──────────────────────────────────────────────┐    │
27
│  │         Semantic Memory                      │    │
28
│  │  - 概念关系                                  │    │
29
│  │  - 知识图谱                                  │    │
30
│  │  - 领域知识                                  │    │
31
│  └──────────────────────────────────────────────┘    │
32
│                                                      │
33
└─────────────────────────────────────────────────────┘

1
class ShortTermMemory:
2
    """短期记忆"""
3

4
    def __init__(self, max_tokens=4000):
5
        self.messages = []
6
        self.max_tokens = max_tokens
7

8
    def add(self, role, content):
9
        """添加消息"""
10
        msg = {'role': role, 'content': content}
11
        self.messages.append(msg)
12
        self._truncate()
13

14
    def get_messages(self):
15
        return self.messages
16

17
    def _truncate(self):
18
        """超出 token 限制时截断"""
19
        while self._count_tokens() > self.max_tokens:
20
            # 保留 system prompt
21
            if len(self.messages) > 2:
22
                self.messages.pop(1)  # 移除第二条 (最早用户消息)
23

24
    def _count_tokens(self):
25
        return sum(len(m['content']) // 4 for m in self.messages)

1
import numpy as np
2
from typing import List, Tuple
3

4
class LongTermMemory:
5
    """长期记忆 - 向量存储"""
6

7
    def __init__(self, embedder, vector_store):
8
        self.embedder = embedder  # embedding 模型
9
        self.vector_store = vector_store  # 向量数据库
10

11
    def store(self, content, metadata=None):
12
        """存储记忆"""
13
        embedding = self.embedder.embed(content)
14
        self.vector_store.add(
15
            vector=embedding,
16
            content=content,
17
            metadata=metadata or {}
18
        )
19

20
    def retrieve(self, query, top_k=5):
21
        """检索相关记忆"""
22
        query_embedding = self.embedder.embed(query)
23
        results = self.vector_store.search(
24
            vector=query_embedding,
25
            top_k=top_k
26
        )
27
        return results
28

29
    def retrieve_relevant_for_context(self, query):
30
        """检索并格式化为上下文"""
31
        results = self.retrieve(query)
32
        context = "\n".join([
33
            f"[{i}] {r['content']}"
34
            for i, r in enumerate(results)
35
        ])
36
        return context

1
class MemoryCompressor:
2
    """记忆压缩器"""
3

4
    def __init__(self, llm):
5
        self.llm = llm
6

7
    def summarize_conversation(self, messages):
8
        """总结对话"""
9
        conversation = "\n".join([
10
            f"{m['role']}: {m['content']}"
11
            for m in messages
12
        ])
13

14
        prompt = f"""
15
        请总结以下对话的关键信息:
16
        {conversation}
17

18
        总结应包含:
19
        1. 用户的主要需求
20
        2. 已完成的工作
21
        3. 重要的决定或发现
22
        4. 待解决的问题
23

24
        总结:
25
        """
26
        return self.llm.generate(prompt)
27

28
    def compress_old_memories(self, memories):
29
        """压缩旧记忆"""
30
        # 相似记忆去重
31
        unique = self._deduplicate(memories)
32

33
        # 过长记忆总结
34
        compressed = []
35
        for memory in unique:
36
            if len(memory) > 500:
37
                summarized = self.summarize_conversation([{'content': memory}])
38
                compressed.append(summarized)
39
            else:
40
                compressed.append(memory)
41

42
        return compressed

1
class MemoryAugmentedAgent:
2
    """带记忆的 Agent"""
3

4
    def __init__(self, llm, tools, memory):
5
        self.llm = llm
6
        self.tools = tools
7
        self.memory = memory
8

9
    def run(self, user_input):
10
        # 1. 检索相关记忆
11
        relevant_memories = self.memory.retrieve_relevant_for_context(user_input)
12

13
        # 2. 增强 prompt
14
        augmented_prompt = f"""
15
        相关历史:
16
        {relevant_memories}
17

18
        当前请求:
19
        {user_input}
20
        """
21

22
        # 3. 生成响应（可调用工具）
23
        response = self._generate(augmented_prompt)
24

25
        # 4. 存储到记忆
26
        self.memory.store(
27
            f"用户: {user_input}\n助手: {response}"
28
        )
29

30
        return response

1
from typing import Callable, Dict, Any, List
2
from dataclasses import dataclass
3
import json
4

5
@dataclass
6
class Tool:
7
    """工具"""
8
    name: str
9
    description: str
10
    func: Callable
11
    schema: Dict[str, Any]
12

13
    def execute(self, **kwargs) -> str:
14
        try:
15
            result = self.func(**kwargs)
16
            return json.dumps({"success": True, "data": result})
17
        except Exception as e:
18
            return json.dumps({"success": False, "error": str(e)})
19

20
    def to_openai_format(self) -> Dict:
21
        return {
22
            "type": "function",
23
            "function": {
24
                "name": self.name,
25
                "description": self.description,
26
                "parameters": self.schema
27
            }
28
        }
29

30
class ToolRegistry:
31
    """工具注册表"""
32

33
    def __init__(self):
34
        self.tools: Dict[str, Tool] = {}
35

36
    def register(self, name=None, description=None):
37
        def decorator(func):
38
            tool_name = name or func.__name__
39
            tool_desc = description or (func.__doc__ or "")
40
            schema = self._infer_schema(func)
41
            tool = Tool(tool_name, tool_desc, func, schema)
42
            self.tools[tool_name] = tool
43
            return func
44
        return decorator
45

46
    def _infer_schema(self, func):
47
        import inspect
48
        sig = inspect.signature(func)
49
        properties = {}
50
        required = []
51

52
        for name, param in sig.parameters.items():
53
            properties[name] = {"type": "string"}
54
            if param.default is param.empty:
55
                required.append(name)
56

57
        return {
58
            "type": "object",
59
            "properties": properties,
60
            "required": required
61
        }
62

63
    def get(self, name: str) -> Tool:
64
        return self.tools.get(name)
65

66
    def to_openai_tools(self) -> List[Dict]:
67
        return [t.to_openai_format() for t in self.tools.values()]

1
class ToolSelector:
2
    """智能工具选择器"""
3

4
    def __init__(self, all_tools, embedding_model):
5
        self.all_tools = all_tools
6
        self.embedding_model = embedding_model
7
        self.tool_embeddings = self._embed_tools()
8

9
    def _embed_tools(self):
10
        embeddings = {}
11
        for tool in self.all_tools:
12
            tool_text = f"{tool.name}: {tool.description}"
13
            embeddings[tool.name] = self.embedding_model.embed(tool_text)
14
        return embeddings
15

16
    def select(self, query, top_k=5):
17
        """根据查询选择相关工具"""
18
        query_emb = self.embedding_model.embed(query)
19

20
        similarities = {}
21
        for name, tool_emb in self.tool_embeddings.items():
22
            sim = self._cosine_similarity(query_emb, tool_emb)
23
            similarities[name] = sim
24

25
        # 返回 top-k
26
        sorted_tools = sorted(
27
            similarities.items(),
28
            key=lambda x: x[1],
29
            reverse=True
30
        )
31

32
        return [name for name, _ in sorted_tools[:top_k]]
33

34
    def _cosine_similarity(self, a, b):
35
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

1
class RobustToolExecutor:
2
    """健壮的工具执行器"""
3

4
    def __init__(self, tools, max_retries=3):
5
        self.tools = tools
6
        self.max_retries = max_retries
7

8
    async def execute(self, name, args):
9
        """异步执行，带重试"""
10
        last_error = None
11

12
        for attempt in range(self.max_retries):
13
            try:
14
                tool = self.tools.get(name)
15
                if not tool:
16
                    return {"error": f"未知工具: {name}"}
17

18
                result = tool.execute(**args)
19

20
                if isinstance(result, dict) and result.get('success'):
21
                    return result
22
                else:
23
                    raise Exception(result.get('error', 'Unknown error'))
24

25
            except Exception as e:
26
                last_error = e
27

28
                # 重试延迟
29
                if attempt < self.max_retries - 1:
30
                    delay = 2 ** attempt  # 指数退避
31
                    await asyncio.sleep(delay)
32

33
        return {"error": f"执行失败 (尝试 {self.max_retries} 次): {last_error}"}