AI 智能体架构全解:从记忆、工具到多智能体协作的硬核实践

from typing import List, Dictclass ShortTermMemory:    def __init__(self, max_tokens: int = 4096):        self.max_tokens = max_tokens        self.history: List[Dict[str, str]] = []    def add(self, role: str, content: str):        self.history.append({”role”: role, ”content”: content})# 简单的 token 计数模拟：字符数 / 4（英文场景约 4 字符/token）        while sum(len(m[”content”]) for m in self.history) / 4 > self.max_tokens:            self.history.pop(0)# 移除最早的消息    def get_history(self) -> List[Dict[str, str]]:        return self.history

from chromadb.config import Settingsimport chromadbclass LongTermMemory:    def __init__(self, collection_name: str = ”agent_memory”):        self.client = chromadb.Client(Settings(anonymized_telemetry=False))        self.collection = self.client.get_or_create_collection(collection_name)    def store(self, doc_id: str, text: str, metadata: dict = None):        ”””存储记忆片段。        生产环境应使用 embedding 模型生成向量，这里以文本直接存储作演示。        ”””        self.collection.add(            ids=[doc_id],            documents=[text],            metadatas=[metadata or {}]        )    def retrieve(self, query: str, n_results: int = 3) -> List[str]:        ”””基于语义相似度检索 top-n 记忆”””        results = self.collection.query(query_texts=[query], n_results=n_results)        return results[”documents”][0] if results and results[”documents”] else []

from pydantic import BaseModelfrom typing import Callable, Anyimport jsonclass ToolParameter(BaseModel):    name: str    type: str    description: str    required: bool = Trueclass ToolDefinition(BaseModel):    name: str    description: str    parameters: list[ToolParameter]class Tool:    def __init__(self, definition: ToolDefinition, func: Callable):        self.definition = definition        self.func = func    def execute(self, **kwargs) -> Any:        return self.func(**kwargs)class ToolRegistry:    def __init__(self):        self.tools: dict[str, Tool] = {}    def register(self, tool: Tool):        self.tools[tool.definition.name] = tool    def get_tool_schemas(self) -> list[dict]:        ”””生成 OpenAI Function Calling 兼容的 tools 描述”””        schemas = []        for tool in self.tools.values():            schema = {                ”name”: tool.definition.name,                ”description”: tool.definition.description,                ”parameters”: {                    ”type”: ”object”,                    ”properties”: {                        p.name: {”type”: p.type, ”description”: p.description}                        for p in tool.definition.parameters                    },                    ”required”: [p.name for p in tool.definition.parameters if p.required]                }            }            schemas.append(schema)        return schemas    def execute(self, name: str, **kwargs) -> Any:        if name not in self.tools:            raise ValueError(f”工具 {name} 未注册”)        return self.tools[name].execute(**kwargs)

def get_weather(city: str) -> str:    ”””模拟天气接口，生产环境替换为真实 API 调用”””# 实际应调用 OpenWeatherMap / 和风天气 等 API    return f”{city} 当前天气：晴朗，25°C，湿度 60%。”weather_tool = Tool(    definition=ToolDefinition(        name=”get_weather”,        description=”查询指定城市的实时天气信息”,        parameters=[ToolParameter(name=”city”, type=”string”, description=”城市名称，例如 Beijing”)]    ),    func=get_weather)registry = ToolRegistry()registry.register(weather_tool)

REACT_PROMPT = ”””你是一个具备工具调用能力的 AI 智能体。请使用以下工具，按指定格式逐步回答问题。可用工具：{tools}回答格式（严格遵守）：Question: 用户的问题Thought: 分析当前状态，思考下一步应采取的行动Action: tool_name(param1=value1, param2=value2)Observation: 工具返回的结果... （可重复 Thought/Action/Observation 多次）Thought: 我已获得足够信息Final Answer: 对用户问题的最终答案开始！Question: {input}{agent_scratchpad}”””

import reimport astdef parse_action(text: str):    ”””从模型输出中提取 Action 指令”””    pattern = r”Action:\s*(.+?)\((.+)\)”    match = re.search(pattern, text)    if not match:        return None, None    tool_name = match.group(1).strip()    params_str = match.group(2).strip()# 生产环境使用 ast.literal_eval 代替 eval，防止代码注入    try:        kwargs = ast.literal_eval(f”dict({params_str})”)    except (ValueError, SyntaxError):        return tool_name, None    return tool_name, kwargsdef parse_final_answer(text: str) -> str:    ”””提取 Final Answer”””    pattern = r”Final Answer:\s*(.*)”    match = re.search(pattern, text, re.DOTALL)    return match.group(1).strip() if match else None

from typing import Anyclass MultiAgentOrchestrator:    def __init__(self, agents: dict[str, Any]):        self.agents = agents# name -> agent 实例    def run_pipeline(self, steps: list[tuple[str, str]], initial_input: str) -> str:        ”””        steps: [(agent_name, task_desc), ...]        每个 agent 需实现 run(task_desc: str, input: str) -> str 方法        ”””        current_input = initial_input        for agent_name, task_desc in steps:            agent = self.agents[agent_name]            result = agent.run(task_desc, current_input)            print(f”[{agent_name}] 输出: {result[:100]}...”)            current_input = result        return current_input

{  ”from”: ”researcher_agent”,  ”to”: ”writer_agent”,  ”type”: ”research_report”,  ”payload”: {    ”summary”: ”2026年AI智能体领域的最新进展包括...”,    ”references”: [”https://arxiv.org/abs/...”, ”https://blog.example.com/...”]  },  ”metadata”: {    ”timestamp”: 1698849600,    ”priority”: ”normal”  }}

class BlogWriterAgent:    def __init__(self, llm, tools_registry, memory):        self.llm = llm        self.tools = tools_registry        self.memory = memory    def run(self, topic: str) -> str:# 步骤 1：检索题目相关背景资料        search_result = self.tools.execute(”web_search”, query=topic)        self.memory.store(”topic_background”, search_result)# 步骤 2：生成大纲        outline_prompt = f”根据以下资料为技术博客《{topic}》生成详细大纲（以减号列表输出）：\n{search_result}”        outline = self.llm.generate(outline_prompt)# 步骤 3：分章节写作并调用工具核实准确性        chapters = [ch.strip(”- ”) for ch in outline.split(”\n”) if ch.strip()]        final_article = []        for chap in chapters:            related_memories = self.memory.retrieve(chap)            memory_context = ”\n”.join(related_memories) if related_memories else ”无相关记忆”            section_prompt = (                f”撰写以下章节的技术博客内容（风格：深入浅出、带代码示例）：\n”                f”章节：{chap}\n”                f”参考资料：\n{memory_context}”            )            draft = self.llm.generate(section_prompt)# 可选：事实核查（如检测引用的论文是否存在、API 名称是否正确）            verification = self.tools.execute(”fact_check”, statement=draft)            if ”错误” in verification:                draft = self.llm.generate(                    f”修正以下内容中的事实错误并保持原有风格：\n{draft}\n注记：{verification}”                )            final_article.append(draft)        return ”\n\n”.join(final_article)