Agent 的工程哲学：简单优于复杂、可观测优于自动化、安全优于速度-开发者社区

Agent 的工程哲学：简单优于复杂、可观测优于自动化、安全优于速度

Agent 是 2025 年最火的技术方向之一，同时也是事故率最高的方向之一。根本原因不是技术不成熟，而是工程哲学不清晰。很多人用做 demo 的思路做 Agent 系统，结果 Agent 在生产环境里像一个喝醉的实习生——有时候很聪明，有时候犯的错让你怀疑人生。

经过过去一年多个 Agent 项目的实战，我提炼了三条 Agent 工程哲学。这三条不是技术建议，是价值排序——在多个目标冲突时，你应该知道优先保什么。

一、深度引言与场景痛点

传统的软件工程哲学是"正确性 > 性能 > 可维护性"。到了 AI 时代，很多团队把这一套直接搬过来，发现不灵了。

原因很简单：Agent 不是一个确定性系统。传统软件的输入和输出之间有明确的映射关系，你可以验证正确性。Agent 的输出不是"对或错"，而是"好、还行、糟"的连续光谱。在一个不确定的系统里追求确定性，就像要求天气预报"必须准确"——道理上没错，工程上做不到。

二、底层机制与原理深度剖析

Agent 项目最容易犯的错误是"过度设计"。你上来就设计了一个五层的 Agent 架构：任务分解 Agent → 工具路由 Agent → 执行 Agent → 验证 Agent → 汇总 Agent。代码写了 5000 行，结果第一个真实查询就卡在任务分解层。

简单不是简陋，是精准。一个三层 Agent（接收 → 规划执行 → 返回）能解决 80% 的任务。剩下的 20% 不值得用 5 倍复杂度去覆盖。

# 复杂的 Agent 架构（不要一开始就这样做） class ComplexAgent: def __init__(self): self.task_decomposer = TaskDecomposer() self.tool_router = ToolRouter() self.executor = Executor() self.validator = Validator() self.summarizer = Summarizer() async def run(self, task): subtasks = await self.task_decomposer.decompose(task) results = [] for sub in subtasks: tool = await self.tool_router.route(sub) result = await self.executor.execute(tool, sub) if await self.validator.validate(result): results.append(result) return await self.summarizer.summarize(results) # 简单的 Agent 架构（从一开始就够用） class SimpleAgent: def __init__(self, model, tools): self.model = model self.tools = tools self.max_steps = 10 async def run(self, task): context = [{"role": "user", "content": task}] for step in range(self.max_steps): response = await self.model.generate(context) tool_call = self._parse_tool_call(response) if not tool_call: return response # 模型认为任务已完成 tool = self.tools.get(tool_call["name"]) if not tool: return f"未知工具: {tool_call['name']}" result = await tool(**tool_call.get("params", {})) context.append({"role": "tool", "content": str(result)}) return "任务步骤超过限制"

简单带来的好处：

出 Bug 时你能在三分钟内定位问题
新同事入职一周就能看懂代码
重构时不用考虑七个组件间的依赖关系
延迟更低（少了五层转发）

简单不是偷懒，是在认识到 Agent 的内在不确定性后，选择用复杂度换可靠性。

三、生产级代码实现

很多人对 Agent 的终极想象是"完全自动化"——Agent 自己思考、自己执行、自己纠错。这个愿景很美，但现在的技术还达不到。

在达不到"完全自动化安全可靠"的阶段，可观测性是比自动化更紧迫的需求。你需要知道：Agent 在做什么、它为什么这么做、它的状态是什么、它在哪一步出错了。

可观测性的四个层次：

第一层：执行日志（最低要求）。记录每一步的模型调用和工具调用：输入什么、输出什么、耗时多少、Token 消耗。

第二层：决策追踪。记录模型做每个决策时的上下文。当 Agent 说"我选择调用订单查询工具"，你要能看到它当时看到了什么信息，让它做出了这个决策。

第三层：状态快照。在关键节点保存完整的 Agent 状态（会话上下文、中间结果、Token 预算余额）。用于复盘时还原现场。

第四层：用户反馈闭环。记录用户对最终答案的反馈（点赞/点踩/沉默），和内部状态关联。分析那些被点踩的对话，找到系统的薄弱环节。

import asyncio import json import time from dataclasses import dataclass, field from typing import Any, Optional from datetime import datetime import logging logger = logging.getLogger(__name__) @dataclass class TraceEvent: event_type: str # "llm_call" | "tool_call" | "decision" | "error" timestamp: str = field( default_factory=lambda: datetime.now().isoformat() ) session_id: str = "" step_number: int = 0 input_summary: str = "" output_summary: str = "" token_count: int = 0 latency_ms: int = 0 metadata: dict = field(default_factory=dict) class ObservableAgent: """带完整可观测性的 Agent 实现""" def __init__(self, model, tools): self.model = model self.tools = tools self.traces: list[TraceEvent] = [] async def run(self, task: str, session_id: str) -> dict: messages = [{"role": "user", "content": task}] step = 0 max_steps = 10 while step < max_steps: step += 1 t_start = time.perf_counter() try: response = await self.model.generate(messages) elapsed = (time.perf_counter() - t_start) * 1000 except Exception as e: self._trace(TraceEvent( event_type="error", session_id=session_id, step_number=step, input_summary=f"LLM调用失败: {e}", latency_ms=int(elapsed) if 'elapsed' in dir() else 0, )) return {"status": "error", "error": str(e)} token_usage = response.get("token_usage", 0) self._trace(TraceEvent( event_type="llm_call", session_id=session_id, step_number=step, input_summary=self._summarize(messages[-1]["content"]), output_summary=self._summarize(response["content"]), token_count=token_usage, latency_ms=int(elapsed), )) tool_call = self._parse_tool_call(response) if not tool_call: return { "status": "success", "answer": response["content"], "steps": step, "token_used": sum( t.token_count for t in self.traces ), "trace_count": len(self.traces), } tool_name = tool_call["name"] params = tool_call.get("params", {}) self._trace(TraceEvent( event_type="decision", session_id=session_id, step_number=step, input_summary=f"选择工具: {tool_name}", metadata={"tool": tool_name, "params": params}, )) tool = self.tools.get(tool_name) if not tool: error_msg = f"未知工具: {tool_name}" self._trace(TraceEvent( event_type="error", session_id=session_id, step_number=step, input_summary=error_msg, )) messages.append({ "role": "tool", "content": f"错误: {error_msg}" }) continue t_start = time.perf_counter() try: result = await tool(**params) elapsed = (time.perf_counter() - t_start) * 1000 except Exception as e: elapsed = (time.perf_counter() - t_start) * 1000 self._trace(TraceEvent( event_type="error", session_id=session_id, step_number=step, input_summary=f"工具 {tool_name} 执行失败: {e}", latency_ms=int(elapsed), )) result = f"工具执行失败: {e}" self._trace(TraceEvent( event_type="tool_call", session_id=session_id, step_number=step, input_summary=f"调用 {tool_name}({json.dumps(params, ensure_ascii=False)})", output_summary=self._summarize(str(result)), latency_ms=int(elapsed), metadata={"tool": tool_name, "params": params}, )) messages.append({ "role": "tool", "tool_call_id": tool_call.get("id", f"call_{step}"), "content": str(result), }) self._trace(TraceEvent( event_type="error", session_id=session_id, step_number=step, input_summary=f"达到最大步数限制 ({max_steps})", )) return { "status": "max_steps_reached", "steps": step, "trace_count": len(self.traces), } def _trace(self, event: TraceEvent): self.traces.append(event) logger.info( f"[{event.session_id}] Step {event.step_number}: " f"{event.event_type} | {event.input_summary[:80]}" ) def _summarize(self, text: str, max_len: int = 100) -> str: if len(text) <= max_len: return text return text[:max_len] + "..." def _parse_tool_call(self, response: dict) -> Optional[dict]: return response.get("tool_calls", [None])[0] def get_trace_summary(self) -> dict: if not self.traces: return {"trace_count": 0} llm_calls = [t for t in self.traces if t.event_type == "llm_call"] tool_calls = [t for t in self.traces if t.event_type == "tool_call"] errors = [t for t in self.traces if t.event_type == "error"] return { "trace_count": len(self.traces), "llm_calls": len(llm_calls), "tool_calls": len(tool_calls), "errors": len(errors), "total_tokens": sum(t.token_count for t in llm_calls), "avg_llm_latency_ms": ( sum(t.latency_ms for t in llm_calls) / len(llm_calls) if llm_calls else 0 ), "avg_tool_latency_ms": ( sum(t.latency_ms for t in tool_calls) / len(tool_calls) if tool_calls else 0 ), }