PythonAI算法
AI Agent 开发进阶：架构、规划、记忆与工具编排
AI Agent 开发进阶涵盖架构设计、规划、记忆与工具编排。文章介绍分层架构（表达、决策、执行、记忆），提供工具注册与函数调用接口实现。包含记忆系统上下文管理、ReAct 循环规划、多代理协作管道及 RAG 知识增强。结合 FastAPI 服务化部署，强调安全治理、日志监控与质量评测。通过完整代码示例展示从零构建工具增强 Agent 的流程，适用于希望落地 Agent 能力的工程师。
AI Agent 开发进阶涵盖架构设计、规划、记忆与工具编排。文章介绍分层架构（表达、决策、执行、记忆），提供工具注册与函数调用接口实现。包含记忆系统上下文管理、ReAct 循环规划、多代理协作管道及 RAG 知识增强。结合 FastAPI 服务化部署，强调安全治理、日志监控与质量评测。通过完整代码示例展示从零构建工具增强 Agent 的流程，适用于希望落地 Agent 能力的工程师。
AI Agent 将大语言模型与外部工具、记忆系统、规划器结合，形成可执行的智能体。它能理解复杂任务、主动调用工具、跨多步达成目标，在研发、数据、客服与自动化场景中显著提升效率与质量。
系统提示词、角色约束与模板管理，决定 Agent 的目标与边界。
规划器与策略选择，决定是否调用工具、如何分解与执行任务。
工具路由与调用、结果解析与持久化，保障事实与行动的可用性。
短期会话上下文与长期知识库，提供跨轮与跨任务的持续性。
各层之间通过清晰接口解耦，支持替换模型、工具或存储实现。
from typing import Dict, Any, Callable
class Tool:
def __init__(self, name: str, description: str, schema: Dict[str, Any], func: Callable[[Dict[str, Any]], Dict[str, Any]]):
self.name = name
self.description = description
self.schema = schema
self.func = func
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
def list(self) -> Dict[str, Dict[str, Any]]:
return {k: v.schema for k, v in self._tools.items()}
def call(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
if name not in self._tools:
return {"error": f"unknown tool {name}"}
return self._tools[name].func(args)
search_tool = Tool(
name="web_search",
description="Search the web",
schema={"type":"object","properties":{"query":{"type":"string"}},"required":["query"]},
func=lambda args: {"results": [f"Result for {args['query']}"]}
)
calc_tool = Tool(
name="calculator",
description="Evaluate arithmetic expression",
schema={"type":"object","properties":{"expr":{"type":"string"}},"required":["expr"]},
func=lambda args: {"value": eval(args["expr"])}
)
registry = ToolRegistry()
registry.register(search_tool)
registry.register(calc_tool)
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")
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"
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])
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]]
针对长对话，采用摘要与优先级保留策略，将事实与决策保留、细枝末节压缩，控制窗口占用。
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
class Decomposer:
def split(self, goal: str) -> list:
parts = goal.split(";")
return [p.strip() for p in parts if p.strip()]
class Aggregator:
def combine(self, results: list) -> str:
return " | ".join(results)
根据任务类型选择不同策略或工具组合，形成策略路由与执行路径。
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
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)
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)
将检索片段与当前上下文合成，供模型或分析员综合输出，降低幻觉与提升覆盖率。
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)}
接收批量目标，分解为子任务并并行或串行执行，提升吞吐与可用性。
记录每步决策、工具调用参数与结果、耗时与错误；统计成功率、重试率与平均延迟，纳入告警。
提示词与参数校验、敏感操作确认、工具白名单与沙箱隔离、审计轨迹与回放。
构建任务集、期望输出与判分规则，自动化回归并人工抽检关键流程。
在核心工具与场景做故障演练与降级方案，保证异常条件下可用性。
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")
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)
context = rag.retrieve("Python features")
print(context)
本文系统讲解了 AI Agent 的分层架构、工具调用与函数接口、记忆系统与向量检索、规划与多步执行、多代理协作、RAG 集成、部署与可观察性、安全治理与评测，并以完整代码组合示例跑通原型。扩展方向包括更强规划器、结构化函数调用协议、知识库平台化与治理、跨系统编排与度量。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online