呕文不踢足球1. 从零构建AI Agent的核心思路
如果你曾经尝试过构建一个AI Agent，很可能遇到过这些问题：它要么陷入无限思考循环，要么像个无头苍蝇一样反复调用工具，或者像个复读机一样不断重复相同内容，更糟的是token消耗速度堪比漏水的管道。这些都不是真正的AI Agent，充其量只是个带着工具的聊天机器人。
真正的AI Agent应该是一个闭环系统：规划→行动（工具调用）→观察→记忆→重新规划→停止。这个循环机制让它能够自主决策下一步行动，而不仅仅是单次响应。关键在于控制而非智能——一个优秀的Agent不是靠"聪明"，而是靠严谨的结构化输出、安全的工具调用、高效的内存管理和可靠的停止条件。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 最小化Agent架构设计
2.1 核心组件分解
我们的最小化Agent包含五个关键子系统：
-
工具系统：每个工具都是独立的Python函数，包含名称、描述、输入JSON模式和执行函数。工具注册中心管理所有可用工具，并提供给模型清晰的工具描述。
-
记忆系统：
- 短期记忆：保存最近18条对话和工具结果
- 长期记忆：持久化存储到JSONL文件的紧凑笔记，支持基于关键词的检索
-
规划系统：
- 高层规划：3-6个要点的任务分解
- 下一步行动：具体到单个工具调用或最终回答
-
停止条件：
- 硬限制：最大步数、最大工具调用次数、最长时间
- 软限制：无进展检测、计划稳定性判断
-
LLM交互：直接调用OpenAI API，强制结构化JSON输出
2.2 控制流设计
Agent的核心工作流程如下：
- 初始化时加载工具、记忆和配置
- 接收用户目标后进入循环：
a. 检查停止条件
b. 构建包含记忆和当前状态的提示
c. 获取模型决策（强制JSON输出）
d. 执行工具调用或存储记忆
e. 更新状态和追踪日志 - 直到满足停止条件或获得最终答案
这个设计的关键在于每次迭代都严格限制模型只能输出预定义结构的JSON，从根本上避免了自由文本导致的不可预测行为。
3. 核心代码实现解析
3.1 工具系统实现
工具注册中心使用装饰器模式，每个工具都是独立的Python函数：
python复制@dataclass
class Tool:
name: str
description: str
schema: Dict[str, Any] # JSON Schema
fn: Callable[[Dict[str, Any]], Any]
safe: bool = True # 危险工具需要特别授权
class ToolRegistry:
def __init__(self):
self._tools = {}
def register(self, tool: Tool):
if tool.name in self._tools:
raise ValueError(f"重复的工具名称: {tool.name}")
self._tools[tool.name] = tool
def get(self, name: str) -> Optional[Tool]:
return self._tools.get(name)
示例工具实现（安全计算器）：
python复制def tool_calc(args: Dict[str, Any]) -> Any:
expr = str(args.get("expression", "")).strip()
if not re.fullmatch(r"[0-9\.\+\-\*\/\(\)\s]+", expr):
raise ValueError("表达式包含不安全字符")
return eval(expr, {"__builtins__": {}}, {})
3.2 记忆系统实现
长期记忆使用JSONL文件存储，支持简单的关键词检索：
python复制class LongTermMemory:
def __init__(self, path: str = "agent_memory.jsonl"):
self.path = path
if not os.path.exists(self.path):
open(self.path, "w").close()
def add(self, text: str, tags: List[str] = None):
note = {
"ts": int(time.time()),
"text": text.strip(),
"tags": tags or []
}
with open(self.path, "a") as f:
f.write(json.dumps(note) + "\n")
def search(self, query: str, k: int = 5) -> List[Dict]:
query_words = set(query.lower().split())
results = []
with open(self.path, "r") as f:
for line in f:
try:
note = json.loads(line)
note_words = set(note["text"].lower().split())
score = len(query_words & note_words)
if score > 0:
results.append((score, note))
except:
continue
return [note for _, note in sorted(results, reverse=True)[:k]]
3.3 Agent核心循环
python复制class ScratchAgent:
def run(self, user_goal: str) -> str:
state = AgentState()
self.stm.add("user", user_goal)
while True:
# 检查停止条件
if stop_reason := self._should_stop(state):
return f"安全停止: {stop_reason}"
state.step += 1
decision = self._model_step(user_goal, state)
# 处理不同类型的决策
if decision["type"] == "plan":
self._handle_plan(decision, state)
elif decision["type"] == "tool_call":
self._handle_tool_call(decision, state)
else: # final
return decision["answer"]
4. 关键设计决策解析
4.1 强制结构化输出
系统提示中严格要求模型必须返回特定格式的JSON：
json复制{
"type": "plan" | "tool_call" | "final",
"plan": ["..."], // 可选，type=plan时必需
"next": "...", // 可选，type=plan时必需
"tool": "tool_name", // type=tool_call时必需
"args": {...}, // type=tool_call时必需
"answer": "...", // type=final时必需
"memory_note": "..." // 可选
}
这种设计消除了80%的Agent混乱问题，因为模型不能自由发挥，必须在预定轨道上运行。
4.2 工具安全调用机制
工具调用采用沙箱原则：
- 每个工具必须预先注册，包含完整的输入模式描述
- 工具执行前后有严格的参数验证
- 危险工具(如文件访问)默认禁用，需要显式配置开启
- 所有工具调用都有错误处理和超时保护
python复制def _run_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
tool = self.tools.get(name)
if not tool:
return {"ok": False, "error": f"未知工具: {name}"}
if not tool.safe and not self.cfg.allow_restricted_tools:
return {"ok": False, "error": f"受限工具: {name}"}
try:
result = tool.fn(args)
return {"ok": True, "data": result}
except Exception as e:
return {"ok": False, "error": str(e)}
4.3 记忆压缩策略
为防止上下文膨胀，采用以下策略：
- 短期记忆只保留最近18条消息
- 每条消息内容截断到260字符
- 长期记忆存储时自动提取关键词
- 记忆检索时只返回最相关的5条记录
python复制def _compact_memory_summary(self) -> str:
tail = self.stm.items[-8:]
return "\n".join(
f"{it['role']}: {it['content'][:260]}"
for it in tail
)
5. 生产环境优化建议
5.1 性能优化技巧
-
记忆压缩：每3-5步生成对话摘要替换原始记录
python复制def summarize_conversation(history: List[str]) -> str: prompt = f"""请用3-5点总结以下对话的核心信息： {json.dumps(history, ensure_ascii=False)} """ return llm.chat([{"role": "user", "content": prompt}]) -
缓存机制：对常见工具调用结果进行缓存
python复制from functools import lru_cache @lru_cache(maxsize=100) def cached_calculation(expr: str) -> float: return tool_calc({"expression": expr}) -
并行工具调用：对无依赖的工具调用使用多线程
python复制from concurrent.futures import ThreadPoolExecutor def parallel_tool_call(tools: List[Tuple[str, dict]]): with ThreadPoolExecutor() as executor: futures = [ executor.submit(self._run_tool, name, args) for name, args in tools ] return [f.result() for f in futures]
5.2 可靠性增强
-
反射机制：最终答案生成后增加校验步骤
python复制def reflect_on_answer(answer: str, trace: List[dict]) -> str: prompt = f"""请检查以下回答的潜在问题并改进： 原始回答: {answer} 执行轨迹: {json.dumps(trace, indent=2)} """ return llm.chat([{"role": "user", "content": prompt}]) -
心跳检测：长时间运行的任务定期报告进度
python复制def heartbeat_monitor(): last_active = time.time() while True: if time.time() - last_active > 30: raise TimeoutError("Agent无响应") time.sleep(5) -
回滚机制：失败时恢复到上一个稳定状态
python复制class AgentState: def create_snapshot(self) -> dict: return deepcopy(self.__dict__) def restore_snapshot(self, snapshot: dict): self.__dict__.update(snapshot)
6. 常见问题与调试技巧
6.1 典型问题排查表
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| Agent陷入循环 | 停止条件太宽松/记忆未更新 | 增加循环检测机制，确保每次迭代都更新记忆 |
| 工具调用失败 | 参数不符合schema/工具未注册 | 在调用前验证参数，检查工具注册表 |
| Token消耗过快 | 上下文膨胀/未压缩记忆 | 实现记忆摘要功能，限制历史消息长度 |
| 响应时间过长 | 复杂工具调用/网络延迟 | 设置超时机制，考虑异步调用 |
6.2 调试日志分析
Agent的每个步骤都会生成详细的追踪日志：
json复制{
"step": 3,
"decision": {
"type": "tool_call",
"tool": "search_web",
"args": {"query": "Python Agent框架对比"}
},
"tool_result": {
"ok": true,
"data": "...",
"latency_ms": 1245
},
"memory_update": {
"short_term": "新增1条",
"long_term": "新增0条"
}
}
分析这些日志时重点关注：
- 决策类型分布是否合理
- 工具调用成功率
- 各步骤耗时分布
- 记忆使用效率
6.3 性能调优参数
关键可调参数及其影响：
python复制class AgentConfig:
max_steps: int = 10 # 最大迭代次数
max_tool_calls: int = 6 # 最大工具调用次数
max_seconds: int = 30 # 最大运行时间(秒)
stm_capacity: int = 18 # 短期记忆容量
ltm_search_limit: int = 5 # 长期记忆检索条数
temperature: float = 0.2 # LLM温度参数
调整原则：
- 从严格限制开始，逐步放宽
- 监控实际使用情况设置合理阈值
- 不同任务类型使用不同配置预设
7. 扩展方向与进阶改造
7.1 多Agent协作系统
基于现有Agent构建协同系统：
python复制class Coordinator:
def __init__(self, agents: List[ScratchAgent]):
self.agents = agents
self.task_queue = Queue()
def dispatch_task(self, task: dict):
expert = self._select_agent(task)
future = expert.run(task["goal"])
self._monitor_progress(future)
def _select_agent(self, task: dict) -> ScratchAgent:
# 根据任务类型选择最合适的Agent
...
7.2 支持自定义工具
允许运行时动态添加工具：
python复制def register_new_tool(agent: ScratchAgent, tool_def: dict):
def tool_func(args):
# 动态生成的工具函数
...
tool = Tool(
name=tool_def["name"],
description=tool_def["description"],
schema=tool_def["schema"],
fn=tool_func
)
agent.tools.register(tool)
7.3 可视化监控界面
使用Web界面展示Agent运行状态：
python复制from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/agent/status")
def agent_status():
return jsonify({
"steps": agent.state.step,
"tool_calls": agent.state.tool_calls,
"memory": {
"stm": len(agent.stm.items),
"ltm": count_ltm_entries()
}
})
这个最小化Agent实现虽然只有约500行代码，但包含了构建可靠AI Agent系统的所有核心要素。它避免了主流框架的复杂性，让开发者能够真正理解Agent工作的每个细节。正如我在实际项目中使用这个架构的经验表明，这种"从零开始"的方法不仅能帮助快速原型开发，也能为后续扩展打下坚实基础。