AI Agent 开发进阶：架构、规划、记忆与工具编排

3.2 函数调用协议

3.2.1 模型输出结构与解析

from typing import Optional

class ModelOutput:
    def __init__(self, content: str, tool_name: Optional[str] = None, tool_args: Optional[Dict[str, Any]] = None):
        self.content = content
        self.tool_name = tool_name
        self.tool_args = tool_args

class SimpleDecider:
    def decide(self, prompt: str) -> ModelOutput:
        if "search:" in prompt:
            q = prompt.split("search:")[-1].strip()
            return ModelOutput(content="use_tool", tool_name="web_search", tool_args={"query": q})
        if "calc:" in prompt:
            e = prompt.split("calc:")[-1].strip()
            return ModelOutput(content="use_tool", tool_name="calculator", tool_args={"expr": e})
        return ModelOutput(content="answer")

3.2.2 决策循环与工具执行

class Executor:
    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def step(self, decision: ModelOutput) -> str:
        if decision.tool_name:
            result = self.registry.call(decision.tool_name, decision.tool_args or {})
            return str(result)
        return "no_tool"

4. 记忆系统与上下文管理

4.1 会话与短期记忆

4.1.1 会话窗口管理

class Memory:
    def __init__(self, max_turns: int = 12):
        self.turns = []
        self.max_turns = max_turns

    def add(self, role: str, content: str):
        self.turns.append({"role": role, "content": content})
        if len(self.turns) > self.max_turns:
            self.turns = self.turns[-self.max_turns:]

    def to_prompt(self) -> str:
        return "\n".join([f"{t['role']}: {t['content']}" for t in self.turns])

4.2 长期记忆与向量检索

4.2.1 轻量向量索引

import math

class VectorStore:
    def __init__(self):
        self.items = []

    def embed(self, text: str) -> list:
        return [
            float(len(text) % 7),
            float(sum(ord(c) for c in text) % 11),
            float(text.count(' '))
        ]

    def add(self, text: str):
        self.items.append((text, self.embed(text)))

    def search(self, query: str, top_k: int = 3):
        q = self.embed(query)
        scored = []
        for t, v in self.items:
            dot = sum(a * b for a, b in zip(q, v))
            qa = math.sqrt(sum(a * a for a in q))
            va = math.sqrt(sum(a * a for a in v))
            s = dot / (qa * va + 1e-9)
            scored.append((s, t))
        scored.sort(reverse=True)
        return [t for _, t in scored[:top_k]]

4.3 记忆策略

4.3.1 摘要与优先级

针对长对话，采用摘要与优先级保留策略，将事实与决策保留、细枝末节压缩，控制窗口占用。

5. 规划与多步执行

5.1 ReAct 循环

5.1.1 计划与执行

class PlannerExecutor:
    def __init__(self, decider: SimpleDecider, registry: ToolRegistry, memory: Memory):
        self.decider = decider
        self.registry = registry
        self.memory = memory

    def run(self, goal: str, max_steps: int = 6) -> str:
        self.memory.add("system", "You are a helpful agent.")
        self.memory.add("user", goal)
        for _ in range(max_steps):
            decision = self.decider.decide(self.memory.to_prompt())
            if decision.tool_name:
                result = self.registry.call(decision.tool_name, decision.tool_args or {})
                self.memory.add("tool", str(result))
                continue
        return "Final: " + goal

5.2 任务分解与合并

5.2.1 简易分解器

class Decomposer:
    def split(self, goal: str) -> list:
        parts = goal.split(";")
        return [p.strip() for p in parts if p.strip()]

5.2.2 汇总器

class Aggregator:
    def combine(self, results: list) -> str:
        return " | ".join(results)

5.3 策略选择与路由

5.3.1 路由器

根据任务类型选择不同策略或工具组合，形成策略路由与执行路径。

6. 多代理协作与角色分工

6.1 角色

6.1.1 搜索员与分析员

class SearchAgent:
    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    def act(self, query: str) -> str:
        return str(self.registry.call("web_search", {"query": query}))

class AnalystAgent:
    def synthesize(self, facts: str) -> str:
        return "Summary: " + facts

6.2 协作管道

6.2.1 管道执行

class Pipeline:
    def __init__(self, searcher: SearchAgent, analyst: AnalystAgent):
        self.searcher = searcher
        self.analyst = analyst

    def run(self, query: str) -> str:
        facts = self.searcher.act(query)
        return self.analyst.synthesize(facts)

7. RAG 集成与知识增强

7.1 文档摄取与切片

7.1.1 切片与索引

class RAG:
    def __init__(self, store: VectorStore):
        self.store = store

    def ingest(self, text: str, chunk_size: int = 200):
        chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
        for c in chunks:
            self.store.add(c)

    def retrieve(self, query: str) -> list:
        return self.store.search(query, top_k=3)

7.2 检索与合成

7.2.1 组合输出

将检索片段与当前上下文合成，供模型或分析员综合输出，降低幻觉与提升覆盖率。

8. 部署与接口

8.1 FastAPI 服务化

8.1.1 运行接口

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
memory = Memory()
planner = PlannerExecutor(SimpleDecider(), registry, memory)

class GoalReq(BaseModel):
    goal: str

@app.post("/run")
async def run(req: GoalReq):
    return {"result": planner.run(req.goal)}

8.2 任务编排与批处理

8.2.1 批量执行

接收批量目标，分解为子任务并并行或串行执行，提升吞吐与可用性。

9. 可观察性与安全治理

9.1 日志与指标

9.1.1 建议

记录每步决策、工具调用参数与结果、耗时与错误；统计成功率、重试率与平均延迟，纳入告警。

9.2 安全与权限

9.2.1 策略

提示词与参数校验、敏感操作确认、工具白名单与沙箱隔离、审计轨迹与回放。

10. 评测与质量保障

10.1 任务集与基准

10.1.1 评测框架

构建任务集、期望输出与判分规则，自动化回归并人工抽检关键流程。

10.2 鲁棒性与演练

10.2.1 建议

在核心工具与场景做故障演练与降级方案，保证异常条件下可用性。

11. 组合示例：从零到跑通的工具增强 Agent

11.1 初始化与依赖

11.1.1 组件初始化

registry = ToolRegistry()
registry.register(search_tool)
registry.register(calc_tool)
memory = Memory(max_turns=12)
planner = PlannerExecutor(SimpleDecider(), registry, memory)
store = VectorStore()
rag = RAG(store)
rag.ingest("Python release notes and docs content")

11.2 运行与协作

11.2.1 示例执行

goal = "search: Python 3.12 features; calc: 1+2*3"
decomposer = Decomposer()
steps = decomposer.split(goal)
results = []
for s in steps:
    memory.add("user", s)
    results.append(planner.run(s, max_steps=3))
agg = Aggregator().combine(results)
print(agg)

11.3 RAG 增强查询

11.3.1 检索片段

context = rag.retrieve("Python features")
print(context)

12. 总结与扩展

12.1 知识点回顾与扩展

本文系统讲解了 AI Agent 的分层架构、工具调用与函数接口、记忆系统与向量检索、规划与多步执行、多代理协作、RAG 集成、部署与可观察性、安全治理与评测，并以完整代码组合示例跑通原型。扩展方向包括更强规划器、结构化函数调用协议、知识库平台化与治理、跨系统编排与度量。

12.2 更多阅读资料

• ReAct: Synergizing Reasoning and Acting https://arxiv.org/abs/2210.03629
• Toolformer https://arxiv.org/abs/2302.04761
• LangChain 文档 https://python.langchain.com
• OpenAI Function Calling 指南 https://platform.openai.com/docs/guides/function-calling
• FastAPI 文档 https://fastapi.tiangolo.com

12.3 新问题与其它方案

• 如何统一工具协议与安全沙箱以降低风险并提升复用性。
• 是否需要多代理协作与角色治理来提升复杂任务完成率。
• 在强检索场景下，RAG 与规划器协作如何平衡成本与质量。
• 如何构建可重复、可量化的评测体系并纳入 CI/CD。