作者的话从ChatGPT到AutoGPTAI Agent正在成为大模型应用的新范式。不同于传统的问答系统AI Agent具备规划、记忆、工具使用等能力可以自主完成复杂任务。本文将深入解析AI Agent的架构设计并带你从零构建一个完整的智能体系统一、什么是AI Agent1.1 Agent vs 传统AI传统AI如ChatGPT被动响应用户输入单次对话无状态保持无法执行外部操作AI Agent主动规划和执行维护长期记忆调用工具搜索、代码执行、API等自主决策迭代优化传统AI交互 AI Agent工作流 ↓ ↓ 用户提问 → AI回答 目标输入 → Agent规划 ↓ 执行工具调用 ↓ 观察结果 ↓ 反思调整 ↓ 输出结果1.2 Agent的核心能力AI Agent架构 ┌─────────────────────────────────────────────┐ │ 规划 (Planning) │ │ 任务分解 → 步骤规划 → 策略选择 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 记忆 (Memory) │ │ 短期记忆对话历史 │ │ 长期记忆知识库、经验 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 工具 (Tools) │ │ 搜索引擎、代码执行、API调用、数据库 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 行动 (Action) │ │ 执行决策、输出结果 │ └─────────────────────────────────────────────┘1.3 ReAct框架ReActReasoning Acting是构建Agent的经典框架ReAct循环 Thought思考→ Action行动→ Observation观察→ ... 示例 Question: 苹果公司2024年的营收是多少 Thought: 我需要搜索苹果公司2024年的财务报告 Action: Search[苹果公司 2024年 营收 财务报告] Observation: 搜索结果显示... Thought: 根据搜索结果我找到了营收数据 Action: Finish[苹果公司2024年营收为XX亿美元]二、Agent核心组件实现2.1 基础Agent类from typing import List, Dict, Any, Callable from dataclasses import dataclass import json dataclass class Action: Agent行动 name: str args: Dict[str, Any] dataclass class Observation: 观察结果 content: str class BaseAgent: 基础Agent类 def __init__(self, llm_client, tools: Dict[str, Callable]): self.llm llm_client self.tools tools self.memory [] # 对话历史 self.max_iterations 10 def run(self, query: str) - str: 运行Agent完成任务 self.memory [{role: user, content: query}] for i in range(self.max_iterations): # 1. 思考下一步 thought self.think() # 2. 决定行动 action self.decide_action(thought) # 3. 执行行动 if action.name Finish: return action.args.get(answer, ) observation self.execute(action) # 4. 更新记忆 self.update_memory(thought, action, observation) return 达到最大迭代次数任务未完成 def think(self) - str: 思考下一步 prompt self._build_prompt() response self.llm.complete(prompt) return response def decide_action(self, thought: str) - Action: 根据思考决定行动 # 解析LLM输出提取Action # 格式: Action: tool_name[arg1value1, arg2value2] pass def execute(self, action: Action) - Observation: 执行工具调用 if action.name not in self.tools: return Observation(f错误: 未知工具 {action.name}) tool self.tools[action.name] result tool(**action.args) return Observation(str(result)) def update_memory(self, thought: str, action: Action, obs: Observation): 更新记忆 self.memory.append({ role: assistant, content: fThought: {thought} Action: {action.name}({action.args}) Observation: {obs.content} })2.2 工具定义# 工具实现示例 import requests from bs4 import BeautifulSoup def search_tool(query: str) - str: 搜索引擎工具 # 实际实现可调用SerpAPI、Bing API等 url fhttps://www.googleapis.com/customsearch/v1 params { key: YOUR_API_KEY, cx: YOUR_SEARCH_ENGINE_ID, q: query } response requests.get(url, paramsparams) data response.json() results [] for item in data.get(items, [])[:3]: results.append(f{item[title]}: {item[snippet]}) return .join(results) def calculator_tool(expression: str) - str: 计算器工具 try: # 安全计算 allowed_names {abs: abs, max: max, min: min, sum: sum} result eval(expression, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {str(e)} def code_executor_tool(code: str, language: str python) - str: 代码执行工具 import subprocess import tempfile import os with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code) temp_file f.name try: result subprocess.run( [python, temp_file], capture_outputTrue, textTrue, timeout10 ) output result.stdout if result.returncode 0 else result.stderr return output finally: os.unlink(temp_file) # 工具注册 tools { Search: search_tool, Calculator: calculator_tool, CodeExecutor: code_executor_tool }2.3 完整ReAct Agentimport re class ReActAgent(BaseAgent): ReAct Agent实现 def __init__(self, llm_client, tools: Dict[str, Callable]): super().__init__(llm_client, tools) self.system_prompt 你是一个智能助手可以使用以下工具完成任务 可用工具 {tool_descriptions} 请使用以下格式思考 Thought: 你的思考过程 Action: 工具名称[参数1值1, 参数2值2] Observation: 工具执行结果 当你认为任务完成时 Thought: 我认为已经找到答案 Action: Finish[answer最终答案] def _build_prompt(self) - str: 构建提示词 tool_desc .join([ f- {name}: {func.__doc__} for name, func in self.tools.items() ]) prompt self.system_prompt.format(tool_descriptionstool_desc) prompt 对话历史 for msg in self.memory: role 用户 if msg[role] user else 助手 prompt f{role}: {msg[content]} prompt Thought: return prompt def decide_action(self, thought: str) - Action: 解析行动 # 提取Action行 action_match re.search(rAction:s*(w)[(.*?)], thought) if action_match: tool_name action_match.group(1) args_str action_match.group(2) # 解析参数 args {} for pair in args_str.split(,): if in pair: key, value pair.split(, 1) args[key.strip()] value.strip() return Action(tool_name, args) # 默认返回思考内容 return Action(Think, {content: thought}) def run(self, query: str) - str: 运行Agent print(f 任务: {query} ) self.memory [{role: user, content: query}] for i in range(self.max_iterations): print(f 迭代 {i1}/{self.max_iterations}) # 思考 thought self.think() print(f Thought: {thought[:200]}...) # 决定行动 action self.decide_action(thought) print(f Action: {action.name}({action.args})) # 检查是否完成 if action.name Finish: answer action.args.get(answer, ) print(f ✅ 任务完成) print(f 答案: {answer}) return answer # 执行 observation self.execute(action) print(f Observation: {observation.content[:200]}...) # 更新记忆 self.update_memory(thought, action, observation) return ❌ 达到最大迭代次数任务未完成三、记忆系统设计3.1 短期记忆class ShortTermMemory: 短期记忆维护对话历史 def __init__(self, max_turns: int 10): self.history [] self.max_turns max_turns def add(self, role: str, content: str): 添加记忆 self.history.append({ role: role, content: content, timestamp: time.time() }) # 保持最近N轮对话 if len(self.history) self.max_turns * 2: self.history self.history[-self.max_turns * 2:] def get_context(self) - str: 获取对话上下文 return .join([ f{msg[role]}: {msg[content]} for msg in self.history ]) def clear(self): 清空记忆 self.history []3.2 长期记忆from sentence_transformers import SentenceTransformer import chromadb import numpy as np class LongTermMemory: 长期记忆基于向量数据库的知识存储 def __init__(self, collection_name: str agent_memory): self.encoder SentenceTransformer(all-MiniLM-L6-v2) self.client chromadb.Client() # 创建或获取集合 try: self.collection self.client.create_collection(namecollection_name) except: self.collection self.client.get_collection(namecollection_name) def add(self, content: str, metadata: Dict None): 添加记忆 embedding self.encoder.encode(content).tolist() doc_id str(hash(content)) self.collection.add( embeddings[embedding], documents[content], metadatas[metadata or {}], ids[doc_id] ) def search(self, query: str, top_k: int 3) - List[str]: 检索相关记忆 query_embedding self.encoder.encode(query).tolist() results self.collection.query( query_embeddings[query_embedding], n_resultstop_k ) return results[documents][0] if results[documents] else [] def get_relevant_context(self, query: str) - str: 获取与查询相关的上下文 memories self.search(query, top_k5) if memories: return 相关背景知识 .join( f- {mem} for mem in memories ) return class MemoryManager: 记忆管理器整合短期和长期记忆 def __init__(self): self.short_term ShortTermMemory() self.long_term LongTermMemory() def add_interaction(self, query: str, response: str, importance: float 0.5): 添加交互到记忆 # 添加到短期记忆 self.short_term.add(user, query) self.short_term.add(assistant, response) # 重要信息存入长期记忆 if importance 0.7: self.long_term.add( fQ: {query} A: {response}, metadata{type: important_qa} ) def build_context(self, current_query: str) - str: 构建完整上下文 context_parts [] # 添加长期记忆 relevant self.long_term.get_relevant_context(current_query) if relevant: context_parts.append(relevant) # 添加短期记忆 context_parts.append(最近对话) context_parts.append(self.short_term.get_context()) return .join(context_parts)四、规划与任务分解4.1 任务分解dataclass class Task: 子任务 id: str description: str dependencies: List[str] # 依赖的其他任务ID status: str pending # pending, in_progress, completed, failed result: Any None class TaskPlanner: 任务规划器 def __init__(self, llm_client): self.llm llm_client self.tasks {} def decompose(self, goal: str) - List[Task]: 将目标分解为子任务 prompt f将以下目标分解为具体的子任务列表 目标: {goal} 请以JSON格式输出子任务列表每个任务包含id、description和dependencies {{ tasks: [ {{id: 1, description: 子任务1, dependencies: []}}, {{id: 2, description: 子任务2, dependencies: [1]}} ] }} response self.llm.complete(prompt) # 解析JSON try: data json.loads(response) tasks [ Task( idt[id], descriptiont[description], dependenciest.get(dependencies, []) ) for t in data[tasks] ] # 存储任务 for task in tasks: self.tasks[task.id] task return tasks except: # 解析失败返回单个任务 return [Task(id1, descriptiongoal, dependencies[])] def get_executable_tasks(self) - List[Task]: 获取当前可执行的任务依赖已完成 executable [] for task in self.tasks.values(): if task.status pending: # 检查依赖 deps_completed all( self.tasks[dep_id].status completed for dep_id in task.dependencies ) if deps_completed: executable.append(task) return executable def execute_plan(self, goal: str, agent: BaseAgent) - str: 执行完整计划 print(f 目标: {goal}) # 分解任务 tasks self.decompose(goal) print(f 分解为 {len(tasks)} 个子任务) # 执行任务 while True: executable self.get_executable_tasks() if not executable: # 检查是否全部完成 if all(t.status completed for t in self.tasks.values()): return 所有任务完成 elif any(t.status failed for t in self.tasks.values()): return 部分任务失败 else: return 任务依赖出现问题 for task in executable: print(f ▶️ 执行任务: {task.description}) task.status in_progress try: result agent.run(task.description) task.result result task.status completed print(f✅ 完成: {result[:100]}...) except Exception as e: task.status failed print(f❌ 失败: {str(e)})4.2 反思与自我改进class SelfReflectiveAgent(ReActAgent): 具备反思能力的Agent def __init__(self, llm_client, tools: Dict[str, Callable]): super().__init__(llm_client, tools) self.reflections [] def reflect(self) - str: 反思之前的行动 if len(self.memory) 3: return prompt f回顾之前的行动进行反思 对话历史 {self._format_history()} 请回答 1. 之前的行动是否有效 2. 有什么可以改进的地方 3. 下一步应该采取什么策略 反思 reflection self.llm.complete(prompt) self.reflections.append(reflection) return reflection def _format_history(self) - str: 格式化历史记录 return .join([ f{msg[role]}: {msg[content][:200]} for msg in self.memory[-5:] ]) def run(self, query: str) - str: 带反思的运行 print(f 任务: {query} ) self.memory [{role: user, content: query}] for i in range(self.max_iterations): # 每隔几步进行反思 if i 0 and i % 3 0: reflection self.reflect() print(f 反思: {reflection[:200]}...) self.memory.append({role: system, content: f反思: {reflection}}) # 思考 thought self.think() print(f Thought: {thought[:150]}...) # 行动 action self.decide_action(thought) print(f Action: {action.name}) if action.name Finish: return action.args.get(answer, ) # 执行 observation self.execute(action) print(f Observation: {observation.content[:150]}...) self.update_memory(thought, action, observation) return 达到最大迭代次数五、完整实战项目5.1 构建个人AI助手# main.py - 完整Agent系统 from agent import ReActAgent, TaskPlanner, MemoryManager from tools import search_tool, calculator_tool, code_executor_tool from llm import OpenAIClient def main(): # 初始化LLM llm OpenAIClient(api_keyyour-api-key) # 定义工具 tools { Search: search_tool, Calculator: calculator_tool, CodeExecutor: code_executor_tool } # 创建Agent agent ReActAgent(llm, tools) # 创建规划器 planner TaskPlanner(llm) print(*60) print( AI Agent助手已启动) print(输入 exit 退出) print(*60) while True: query input( 你: ) if query.lower() exit: print( 再见) break # 简单查询直接回答 if len(query) 50: response agent.run(query) else: # 复杂任务使用规划器 response planner.execute_plan(query, agent) print(f Agent: {response}) if __name__ __main__: main()5.2 多Agent协作系统class MultiAgentSystem: 多Agent协作系统 def __init__(self): self.agents {} self.message_bus [] def register_agent(self, name: str, agent: BaseAgent, role: str): 注册Agent self.agents[name] { agent: agent, role: role } def broadcast(self, sender: str, message: str): 广播消息 self.message_bus.append({ from: sender, content: message, timestamp: time.time() }) def collaborate(self, task: str) - str: 多Agent协作完成任务 # 分配子任务 subtasks self._deallocate_task(task) results {} for agent_name, subtask in subtasks.items(): print(f 分配任务给 {agent_name}: {subtask}) agent self.agents[agent_name][agent] result agent.run(subtask) results[agent_name] result self.broadcast(agent_name, f完成: {result[:100]}) # 整合结果 return self._integrate_results(results) def _deallocate_task(self, task: str) - Dict[str, str]: 任务分配 # 根据Agent角色分配 allocations {} for name, info in self.agents.items(): if info[role] researcher: allocations[name] f搜索相关信息: {task} elif info[role] analyst: allocations[name] f分析数据: {task} elif info[role] writer: allocations[name] f撰写报告: {task} return allocations def _integrate_results(self, results: Dict[str, str]) - str: 整合结果 return .join([ f【{name}】 {result} for name, result in results.items() ])六、总结核心要点Agent架构规划、记忆、工具、行动四大核心组件ReAct框架思考-行动-观察循环实现自主决策记忆系统短期记忆保持对话上下文长期记忆存储知识任务规划复杂任务分解为子任务按依赖顺序执行反思机制自我评估和改进提高任务完成质量应用场景场景能力需求典型应用个人助理工具调用、记忆日程管理、信息查询代码助手代码执行、推理自动编程、调试研究助手搜索、分析、规划文献综述、数据分析客服系统多轮对话、工具调用自动客服、工单处理学习路径Level 1: 基础Agent ├── ReAct框架理解 ├── 工具定义与调用 └── 简单任务执行 Level 2: 增强Agent ├── 记忆系统实现 ├── 任务分解规划 └── 反思机制 Level 3: 多Agent系统 ├── Agent协作 ├── 任务分配 └── 结果整合 Level 4: 生产部署 ├── 错误处理 ├── 安全沙箱 └── 性能优化下一篇预告【第52篇】RAG系统实战检索增强生成技术详解万字长文完整代码实现本文为系列第51篇详细介绍了AI Agent的架构设计和实战实现。有任何问题欢迎在评论区交流标签AI Agent、智能体、ReAct、AutoGPT、LLM应用、工具调用
人工智能【第51篇】AI Agent实战：构建智能体系统
作者的话从ChatGPT到AutoGPTAI Agent正在成为大模型应用的新范式。不同于传统的问答系统AI Agent具备规划、记忆、工具使用等能力可以自主完成复杂任务。本文将深入解析AI Agent的架构设计并带你从零构建一个完整的智能体系统一、什么是AI Agent1.1 Agent vs 传统AI传统AI如ChatGPT被动响应用户输入单次对话无状态保持无法执行外部操作AI Agent主动规划和执行维护长期记忆调用工具搜索、代码执行、API等自主决策迭代优化传统AI交互 AI Agent工作流 ↓ ↓ 用户提问 → AI回答 目标输入 → Agent规划 ↓ 执行工具调用 ↓ 观察结果 ↓ 反思调整 ↓ 输出结果1.2 Agent的核心能力AI Agent架构 ┌─────────────────────────────────────────────┐ │ 规划 (Planning) │ │ 任务分解 → 步骤规划 → 策略选择 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 记忆 (Memory) │ │ 短期记忆对话历史 │ │ 长期记忆知识库、经验 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 工具 (Tools) │ │ 搜索引擎、代码执行、API调用、数据库 │ └─────────────────┬───────────────────────────┘ ↓ ┌─────────────────────────────────────────────┐ │ 行动 (Action) │ │ 执行决策、输出结果 │ └─────────────────────────────────────────────┘1.3 ReAct框架ReActReasoning Acting是构建Agent的经典框架ReAct循环 Thought思考→ Action行动→ Observation观察→ ... 示例 Question: 苹果公司2024年的营收是多少 Thought: 我需要搜索苹果公司2024年的财务报告 Action: Search[苹果公司 2024年 营收 财务报告] Observation: 搜索结果显示... Thought: 根据搜索结果我找到了营收数据 Action: Finish[苹果公司2024年营收为XX亿美元]二、Agent核心组件实现2.1 基础Agent类from typing import List, Dict, Any, Callable from dataclasses import dataclass import json dataclass class Action: Agent行动 name: str args: Dict[str, Any] dataclass class Observation: 观察结果 content: str class BaseAgent: 基础Agent类 def __init__(self, llm_client, tools: Dict[str, Callable]): self.llm llm_client self.tools tools self.memory [] # 对话历史 self.max_iterations 10 def run(self, query: str) - str: 运行Agent完成任务 self.memory [{role: user, content: query}] for i in range(self.max_iterations): # 1. 思考下一步 thought self.think() # 2. 决定行动 action self.decide_action(thought) # 3. 执行行动 if action.name Finish: return action.args.get(answer, ) observation self.execute(action) # 4. 更新记忆 self.update_memory(thought, action, observation) return 达到最大迭代次数任务未完成 def think(self) - str: 思考下一步 prompt self._build_prompt() response self.llm.complete(prompt) return response def decide_action(self, thought: str) - Action: 根据思考决定行动 # 解析LLM输出提取Action # 格式: Action: tool_name[arg1value1, arg2value2] pass def execute(self, action: Action) - Observation: 执行工具调用 if action.name not in self.tools: return Observation(f错误: 未知工具 {action.name}) tool self.tools[action.name] result tool(**action.args) return Observation(str(result)) def update_memory(self, thought: str, action: Action, obs: Observation): 更新记忆 self.memory.append({ role: assistant, content: fThought: {thought} Action: {action.name}({action.args}) Observation: {obs.content} })2.2 工具定义# 工具实现示例 import requests from bs4 import BeautifulSoup def search_tool(query: str) - str: 搜索引擎工具 # 实际实现可调用SerpAPI、Bing API等 url fhttps://www.googleapis.com/customsearch/v1 params { key: YOUR_API_KEY, cx: YOUR_SEARCH_ENGINE_ID, q: query } response requests.get(url, paramsparams) data response.json() results [] for item in data.get(items, [])[:3]: results.append(f{item[title]}: {item[snippet]}) return .join(results) def calculator_tool(expression: str) - str: 计算器工具 try: # 安全计算 allowed_names {abs: abs, max: max, min: min, sum: sum} result eval(expression, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {str(e)} def code_executor_tool(code: str, language: str python) - str: 代码执行工具 import subprocess import tempfile import os with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code) temp_file f.name try: result subprocess.run( [python, temp_file], capture_outputTrue, textTrue, timeout10 ) output result.stdout if result.returncode 0 else result.stderr return output finally: os.unlink(temp_file) # 工具注册 tools { Search: search_tool, Calculator: calculator_tool, CodeExecutor: code_executor_tool }2.3 完整ReAct Agentimport re class ReActAgent(BaseAgent): ReAct Agent实现 def __init__(self, llm_client, tools: Dict[str, Callable]): super().__init__(llm_client, tools) self.system_prompt 你是一个智能助手可以使用以下工具完成任务 可用工具 {tool_descriptions} 请使用以下格式思考 Thought: 你的思考过程 Action: 工具名称[参数1值1, 参数2值2] Observation: 工具执行结果 当你认为任务完成时 Thought: 我认为已经找到答案 Action: Finish[answer最终答案] def _build_prompt(self) - str: 构建提示词 tool_desc .join([ f- {name}: {func.__doc__} for name, func in self.tools.items() ]) prompt self.system_prompt.format(tool_descriptionstool_desc) prompt 对话历史 for msg in self.memory: role 用户 if msg[role] user else 助手 prompt f{role}: {msg[content]} prompt Thought: return prompt def decide_action(self, thought: str) - Action: 解析行动 # 提取Action行 action_match re.search(rAction:s*(w)[(.*?)], thought) if action_match: tool_name action_match.group(1) args_str action_match.group(2) # 解析参数 args {} for pair in args_str.split(,): if in pair: key, value pair.split(, 1) args[key.strip()] value.strip() return Action(tool_name, args) # 默认返回思考内容 return Action(Think, {content: thought}) def run(self, query: str) - str: 运行Agent print(f 任务: {query} ) self.memory [{role: user, content: query}] for i in range(self.max_iterations): print(f 迭代 {i1}/{self.max_iterations}) # 思考 thought self.think() print(f Thought: {thought[:200]}...) # 决定行动 action self.decide_action(thought) print(f Action: {action.name}({action.args})) # 检查是否完成 if action.name Finish: answer action.args.get(answer, ) print(f ✅ 任务完成) print(f 答案: {answer}) return answer # 执行 observation self.execute(action) print(f Observation: {observation.content[:200]}...) # 更新记忆 self.update_memory(thought, action, observation) return ❌ 达到最大迭代次数任务未完成三、记忆系统设计3.1 短期记忆class ShortTermMemory: 短期记忆维护对话历史 def __init__(self, max_turns: int 10): self.history [] self.max_turns max_turns def add(self, role: str, content: str): 添加记忆 self.history.append({ role: role, content: content, timestamp: time.time() }) # 保持最近N轮对话 if len(self.history) self.max_turns * 2: self.history self.history[-self.max_turns * 2:] def get_context(self) - str: 获取对话上下文 return .join([ f{msg[role]}: {msg[content]} for msg in self.history ]) def clear(self): 清空记忆 self.history []3.2 长期记忆from sentence_transformers import SentenceTransformer import chromadb import numpy as np class LongTermMemory: 长期记忆基于向量数据库的知识存储 def __init__(self, collection_name: str agent_memory): self.encoder SentenceTransformer(all-MiniLM-L6-v2) self.client chromadb.Client() # 创建或获取集合 try: self.collection self.client.create_collection(namecollection_name) except: self.collection self.client.get_collection(namecollection_name) def add(self, content: str, metadata: Dict None): 添加记忆 embedding self.encoder.encode(content).tolist() doc_id str(hash(content)) self.collection.add( embeddings[embedding], documents[content], metadatas[metadata or {}], ids[doc_id] ) def search(self, query: str, top_k: int 3) - List[str]: 检索相关记忆 query_embedding self.encoder.encode(query).tolist() results self.collection.query( query_embeddings[query_embedding], n_resultstop_k ) return results[documents][0] if results[documents] else [] def get_relevant_context(self, query: str) - str: 获取与查询相关的上下文 memories self.search(query, top_k5) if memories: return 相关背景知识 .join( f- {mem} for mem in memories ) return class MemoryManager: 记忆管理器整合短期和长期记忆 def __init__(self): self.short_term ShortTermMemory() self.long_term LongTermMemory() def add_interaction(self, query: str, response: str, importance: float 0.5): 添加交互到记忆 # 添加到短期记忆 self.short_term.add(user, query) self.short_term.add(assistant, response) # 重要信息存入长期记忆 if importance 0.7: self.long_term.add( fQ: {query} A: {response}, metadata{type: important_qa} ) def build_context(self, current_query: str) - str: 构建完整上下文 context_parts [] # 添加长期记忆 relevant self.long_term.get_relevant_context(current_query) if relevant: context_parts.append(relevant) # 添加短期记忆 context_parts.append(最近对话) context_parts.append(self.short_term.get_context()) return .join(context_parts)四、规划与任务分解4.1 任务分解dataclass class Task: 子任务 id: str description: str dependencies: List[str] # 依赖的其他任务ID status: str pending # pending, in_progress, completed, failed result: Any None class TaskPlanner: 任务规划器 def __init__(self, llm_client): self.llm llm_client self.tasks {} def decompose(self, goal: str) - List[Task]: 将目标分解为子任务 prompt f将以下目标分解为具体的子任务列表 目标: {goal} 请以JSON格式输出子任务列表每个任务包含id、description和dependencies {{ tasks: [ {{id: 1, description: 子任务1, dependencies: []}}, {{id: 2, description: 子任务2, dependencies: [1]}} ] }} response self.llm.complete(prompt) # 解析JSON try: data json.loads(response) tasks [ Task( idt[id], descriptiont[description], dependenciest.get(dependencies, []) ) for t in data[tasks] ] # 存储任务 for task in tasks: self.tasks[task.id] task return tasks except: # 解析失败返回单个任务 return [Task(id1, descriptiongoal, dependencies[])] def get_executable_tasks(self) - List[Task]: 获取当前可执行的任务依赖已完成 executable [] for task in self.tasks.values(): if task.status pending: # 检查依赖 deps_completed all( self.tasks[dep_id].status completed for dep_id in task.dependencies ) if deps_completed: executable.append(task) return executable def execute_plan(self, goal: str, agent: BaseAgent) - str: 执行完整计划 print(f 目标: {goal}) # 分解任务 tasks self.decompose(goal) print(f 分解为 {len(tasks)} 个子任务) # 执行任务 while True: executable self.get_executable_tasks() if not executable: # 检查是否全部完成 if all(t.status completed for t in self.tasks.values()): return 所有任务完成 elif any(t.status failed for t in self.tasks.values()): return 部分任务失败 else: return 任务依赖出现问题 for task in executable: print(f ▶️ 执行任务: {task.description}) task.status in_progress try: result agent.run(task.description) task.result result task.status completed print(f✅ 完成: {result[:100]}...) except Exception as e: task.status failed print(f❌ 失败: {str(e)})4.2 反思与自我改进class SelfReflectiveAgent(ReActAgent): 具备反思能力的Agent def __init__(self, llm_client, tools: Dict[str, Callable]): super().__init__(llm_client, tools) self.reflections [] def reflect(self) - str: 反思之前的行动 if len(self.memory) 3: return prompt f回顾之前的行动进行反思 对话历史 {self._format_history()} 请回答 1. 之前的行动是否有效 2. 有什么可以改进的地方 3. 下一步应该采取什么策略 反思 reflection self.llm.complete(prompt) self.reflections.append(reflection) return reflection def _format_history(self) - str: 格式化历史记录 return .join([ f{msg[role]}: {msg[content][:200]} for msg in self.memory[-5:] ]) def run(self, query: str) - str: 带反思的运行 print(f 任务: {query} ) self.memory [{role: user, content: query}] for i in range(self.max_iterations): # 每隔几步进行反思 if i 0 and i % 3 0: reflection self.reflect() print(f 反思: {reflection[:200]}...) self.memory.append({role: system, content: f反思: {reflection}}) # 思考 thought self.think() print(f Thought: {thought[:150]}...) # 行动 action self.decide_action(thought) print(f Action: {action.name}) if action.name Finish: return action.args.get(answer, ) # 执行 observation self.execute(action) print(f Observation: {observation.content[:150]}...) self.update_memory(thought, action, observation) return 达到最大迭代次数五、完整实战项目5.1 构建个人AI助手# main.py - 完整Agent系统 from agent import ReActAgent, TaskPlanner, MemoryManager from tools import search_tool, calculator_tool, code_executor_tool from llm import OpenAIClient def main(): # 初始化LLM llm OpenAIClient(api_keyyour-api-key) # 定义工具 tools { Search: search_tool, Calculator: calculator_tool, CodeExecutor: code_executor_tool } # 创建Agent agent ReActAgent(llm, tools) # 创建规划器 planner TaskPlanner(llm) print(*60) print( AI Agent助手已启动) print(输入 exit 退出) print(*60) while True: query input( 你: ) if query.lower() exit: print( 再见) break # 简单查询直接回答 if len(query) 50: response agent.run(query) else: # 复杂任务使用规划器 response planner.execute_plan(query, agent) print(f Agent: {response}) if __name__ __main__: main()5.2 多Agent协作系统class MultiAgentSystem: 多Agent协作系统 def __init__(self): self.agents {} self.message_bus [] def register_agent(self, name: str, agent: BaseAgent, role: str): 注册Agent self.agents[name] { agent: agent, role: role } def broadcast(self, sender: str, message: str): 广播消息 self.message_bus.append({ from: sender, content: message, timestamp: time.time() }) def collaborate(self, task: str) - str: 多Agent协作完成任务 # 分配子任务 subtasks self._deallocate_task(task) results {} for agent_name, subtask in subtasks.items(): print(f 分配任务给 {agent_name}: {subtask}) agent self.agents[agent_name][agent] result agent.run(subtask) results[agent_name] result self.broadcast(agent_name, f完成: {result[:100]}) # 整合结果 return self._integrate_results(results) def _deallocate_task(self, task: str) - Dict[str, str]: 任务分配 # 根据Agent角色分配 allocations {} for name, info in self.agents.items(): if info[role] researcher: allocations[name] f搜索相关信息: {task} elif info[role] analyst: allocations[name] f分析数据: {task} elif info[role] writer: allocations[name] f撰写报告: {task} return allocations def _integrate_results(self, results: Dict[str, str]) - str: 整合结果 return .join([ f【{name}】 {result} for name, result in results.items() ])六、总结核心要点Agent架构规划、记忆、工具、行动四大核心组件ReAct框架思考-行动-观察循环实现自主决策记忆系统短期记忆保持对话上下文长期记忆存储知识任务规划复杂任务分解为子任务按依赖顺序执行反思机制自我评估和改进提高任务完成质量应用场景场景能力需求典型应用个人助理工具调用、记忆日程管理、信息查询代码助手代码执行、推理自动编程、调试研究助手搜索、分析、规划文献综述、数据分析客服系统多轮对话、工具调用自动客服、工单处理学习路径Level 1: 基础Agent ├── ReAct框架理解 ├── 工具定义与调用 └── 简单任务执行 Level 2: 增强Agent ├── 记忆系统实现 ├── 任务分解规划 └── 反思机制 Level 3: 多Agent系统 ├── Agent协作 ├── 任务分配 └── 结果整合 Level 4: 生产部署 ├── 错误处理 ├── 安全沙箱 └── 性能优化下一篇预告【第52篇】RAG系统实战检索增强生成技术详解万字长文完整代码实现本文为系列第51篇详细介绍了AI Agent的架构设计和实战实现。有任何问题欢迎在评论区交流标签AI Agent、智能体、ReAct、AutoGPT、LLM应用、工具调用
