昌都市网站建设_网站建设公司_SQL Server_seo优化

一、知识点解析

Agent 架构是构建智能体系统的核心骨架，包含三个关键维度：

• 单代理 vs 多代理：单代理独立完成任务；多代理通过协作（如分工、辩论、黑板模型）解决复杂问题；
• Agent Loop：主控制循环，持续执行“感知 → 推理 → 行动 → 记忆更新”；
• Memory 机制：存储历史交互（短期记忆）或长期知识（向量/图记忆），支持上下文连贯与跨轮推理。

本教学将实现一个带记忆的单代理 Loop，并扩展为双代理协作架构（Manager + Executor），全部手动编码，清晰展示状态流与通信机制。

二、可运行代码实例

依赖安装

pip install openai python-dotenv

完整代码（单代理 + 双代理 + Memory）

import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # ===== 简易记忆模块（短期对话记忆）===== class SimpleMemory: def __init__(self, max_turns=5): self.history = [] self.max_turns = max_turns def add(self, role: str, content: str): self.history.append({"role": role, "content": content}) if len(self.history) > self.max_turns * 2: # 每轮含 user + assistant self.history = self.history[-self.max_turns * 2:] def get(self): return self.history.copy() # ===== 单代理：带记忆的 ReAct Agent ===== class SingleAgent: def __init__(self, name="Assistant"): self.name = name self.memory = SimpleMemory() def step(self, user_input: str) -> str: self.memory.add("user", user_input) resp = client.chat.completions.create( model="gpt-3.5-turbo", messages=self.memory.get(), temperature=0.0 ) answer = resp.choices[0].message.content.strip() self.memory.add("assistant", answer) return answer # ===== 多代理：Manager + Executor 协作 ===== class ManagerAgent: def __init__(self): self.memory = SimpleMemory() def plan(self, task: str) -> str: prompt = f"Break down the task into one clear subtask for an executor:\nTask: {task}" resp = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.0 ) return resp.choices[0].message.content.strip() class ExecutorAgent: def execute(self, subtask: str) -> str: resp = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": f"Execute: {subtask}"}], temperature=0.0 ) return resp.choices[0].message.content.strip() def multi_agent_run(task: str) -> str: manager = ManagerAgent() executor = ExecutorAgent() subtask = manager.plan(task) result = executor.execute(subtask) # Manager synthesizes final answer final_prompt = f"Task: {task}\nSubtask Result: {result}\nSummarize concisely." resp = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": final_prompt}], temperature=0.0 ) return resp.choices[0].message.content.strip() # ===== 示例运行 ===== if __name__ == "__main__": # 单代理测试（带记忆） agent = SingleAgent() print("[Single Agent]") print(agent.step("What is 2+2?")) print(agent.step("Now multiply that by 3.")) # 应记住前文答案 4 print("\n[Multi Agent]") ans = multi_agent_run("Explain how photosynthesis works in simple terms.") print(ans)

三、代码逐行讲解

▶Memory 机制（第 8–18 行）

• SimpleMemory使用列表存储对话历史，限制最大轮数防止上下文过长；
• add()方法追加消息，get()返回副本供 LLM 调用——这是短期记忆的标准实现；
• 设计要点：不使用全局变量，每个 Agent 拥有独立记忆，避免状态污染。

▶单代理 Loop（第 21–32 行）

• SingleAgent.step()是Agent Loop 的最小单元：接收输入 → 查询记忆 → 调用 LLM → 更新记忆 → 返回输出；
• 关键：messages=self.memory.get()将记忆注入上下文，使 LLM 具备历史感知能力；
• 温度设为 0.0 确保确定性，便于验证记忆是否生效。

▶多代理协作（第 35–60 行）

• ManagerAgent：负责任务分解（规划），体现“思考”角色；
• ExecutorAgent：负责执行子任务，体现“行动”角色；
• multi_agent_run()实现两阶段流水线：Plan → Execute → Synthesize；
• 通信方式：通过函数参数传递子任务和结果（最简化的消息传递）。

▶示例逻辑（第 63–73 行）

• 单代理连续提问验证记忆：“2+2=4” → “4×3=12”；
• 多代理处理复杂任务，由 Manager 拆解，Executor 执行，Manager 整合。

⚠️为什么不用共享内存？
初学者应先掌握显式消息传递（更安全、可追溯），再进阶到共享黑板或向量数据库。

四、效果验证与输出示例

预期输出：

[Single Agent] 4 12 [Multi Agent] Photosynthesis is how plants use sunlight to turn carbon dioxide and water into sugar (food) and oxygen.

关键验证点：

• 单代理第二问能正确引用第一问结果（证明Memory 机制生效）；
• 多代理输出简洁准确（证明分工协作有效）。

💡 若第二问返回“2+2=4, then 4*3=12”，说明记忆成功注入上下文。

五、扩展优化

当前局限性

1. 记忆仅支持对话历史，无长期知识存储；
2. 多代理为静态流水线，无法动态协商；
3. 无错误恢复或重试机制。

优化方向 1：添加长期记忆（向量检索）

# 使用 FAISS + sentence-transformers（需 pip install faiss-cpu sentence-transformers） from sentence_transformers import SentenceTransformer import faiss import numpy as np class VectorMemory: def __init__(self): self.encoder = SentenceTransformer('all-MiniLM-L6-v2') self.index = faiss.IndexFlatL2(384) self.memories = [] def add(self, text: str): emb = self.encoder.encode([text]).astype('float32') self.index.add(emb) self.memories.append(text) def search(self, query: str, k=1) -> list: emb = self.encoder.encode([query]).astype('float32') _, idxs = self.index.search(emb, k) return [self.memories[i] for i in idxs[0]]

优化方向 2：多代理动态协商（简单投票）

# 启动多个 Executor，Manager 投票选择最佳答案 executors = [ExecutorAgent() for _ in range(3)] results = [e.execute(subtask) for e in executors] # Manager 选择最一致的答案（简化版） final = results[0] # 实际可用 LLM 判断一致性