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_factorylambda: 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_factorydict) 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_typeerror, session_idsession_id, step_numberstep, input_summaryfLLM调用失败: {e}, latency_msint(elapsed) if elapsed in dir() else 0, )) return {status: error, error: str(e)} token_usage response.get(token_usage, 0) self._trace(TraceEvent( event_typellm_call, session_idsession_id, step_numberstep, input_summaryself._summarize(messages[-1][content]), output_summaryself._summarize(response[content]), token_counttoken_usage, latency_msint(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_typedecision, session_idsession_id, step_numberstep, input_summaryf选择工具: {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_typeerror, session_idsession_id, step_numberstep, input_summaryerror_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_typeerror, session_idsession_id, step_numberstep, input_summaryf工具 {tool_name} 执行失败: {e}, latency_msint(elapsed), )) result f工具执行失败: {e} self._trace(TraceEvent( event_typetool_call, session_idsession_id, step_numberstep, input_summaryf调用 {tool_name}({json.dumps(params, ensure_asciiFalse)}), output_summaryself._summarize(str(result)), latency_msint(elapsed), metadata{tool: tool_name, params: params}, )) messages.append({ role: tool, tool_call_id: tool_call.get(id, fcall_{step}), content: str(result), }) self._trace(TraceEvent( event_typeerror, session_idsession_id, step_numberstep, input_summaryf达到最大步数限制 ({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 ), }四、边界分析与架构权衡Agent 和传统 API 最大的区别是Agent 的调用链更长、更复杂、更不可预测。传统 API 从请求到响应一般经过 3-5 个组件Agent 可能经过 10-20 次模型调用 工具调用。每一步都是潜在的安全漏洞。安全优先的三个实践最小权限给 Agent 的每个工具只开放它完成任务所需的最小权限。查询订单只需要 SELECT 权限别给 DELETE。读取文件只能读特定目录别给根目录权限。输出校验Agent 的任何输出在返回给用户之前过一遍内容过滤器。过滤 PII电话/邮箱/身份证、有害内容、超出知识范围的猜测。人工确认关键操作发邮件、改配置、执行付费操作在 Agent 执行前必须经过人工确认。不要让 Agent 替你点发送按钮。五、总结简单优于复杂、可观测优于自动化、安全优于速度——这三条哲学背后有一个共同的逻辑在不确定系统中降低认知负担比提升自动化程度更重要。Agent 的内在不确定性决定了你无法像控制传统软件一样控制它。你越是试图用复杂的规则约束它它越容易在你不期望的地方突破约束。与其加更多规则不如让系统本身更简单、更透明、更安全。这三条哲学其实是一种谦虚的工程态度承认 Agent 会犯错所以需要可观测性承认人类比 Agent 更擅长判断所以关键决策需要人工确认承认复杂度会放大错误所以追求简单六、总结这篇文章是这个系列的终章。从第 1 篇的复杂度分析到第 20 篇的工程哲学核心线索只有一条AI 工程不是消灭不确定性而是管理不确定性。Agent 的工程哲学三条简单优于复杂用最少的组件完成最多的任务可观测优于自动化先能看见再谈自动安全优于速度一次安全事故的代价大于所有性能优化的收益记住这三条你做的 Agent 系统不一定会更炫但一定会更稳。而在生产环境里稳 炫。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0730 资料来源索引并在发布前将具体来源贴到对应断言之后。
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_factorylambda: 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_factorydict) 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_typeerror, session_idsession_id, step_numberstep, input_summaryfLLM调用失败: {e}, latency_msint(elapsed) if elapsed in dir() else 0, )) return {status: error, error: str(e)} token_usage response.get(token_usage, 0) self._trace(TraceEvent( event_typellm_call, session_idsession_id, step_numberstep, input_summaryself._summarize(messages[-1][content]), output_summaryself._summarize(response[content]), token_counttoken_usage, latency_msint(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_typedecision, session_idsession_id, step_numberstep, input_summaryf选择工具: {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_typeerror, session_idsession_id, step_numberstep, input_summaryerror_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_typeerror, session_idsession_id, step_numberstep, input_summaryf工具 {tool_name} 执行失败: {e}, latency_msint(elapsed), )) result f工具执行失败: {e} self._trace(TraceEvent( event_typetool_call, session_idsession_id, step_numberstep, input_summaryf调用 {tool_name}({json.dumps(params, ensure_asciiFalse)}), output_summaryself._summarize(str(result)), latency_msint(elapsed), metadata{tool: tool_name, params: params}, )) messages.append({ role: tool, tool_call_id: tool_call.get(id, fcall_{step}), content: str(result), }) self._trace(TraceEvent( event_typeerror, session_idsession_id, step_numberstep, input_summaryf达到最大步数限制 ({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 ), }四、边界分析与架构权衡Agent 和传统 API 最大的区别是Agent 的调用链更长、更复杂、更不可预测。传统 API 从请求到响应一般经过 3-5 个组件Agent 可能经过 10-20 次模型调用 工具调用。每一步都是潜在的安全漏洞。安全优先的三个实践最小权限给 Agent 的每个工具只开放它完成任务所需的最小权限。查询订单只需要 SELECT 权限别给 DELETE。读取文件只能读特定目录别给根目录权限。输出校验Agent 的任何输出在返回给用户之前过一遍内容过滤器。过滤 PII电话/邮箱/身份证、有害内容、超出知识范围的猜测。人工确认关键操作发邮件、改配置、执行付费操作在 Agent 执行前必须经过人工确认。不要让 Agent 替你点发送按钮。五、总结简单优于复杂、可观测优于自动化、安全优于速度——这三条哲学背后有一个共同的逻辑在不确定系统中降低认知负担比提升自动化程度更重要。Agent 的内在不确定性决定了你无法像控制传统软件一样控制它。你越是试图用复杂的规则约束它它越容易在你不期望的地方突破约束。与其加更多规则不如让系统本身更简单、更透明、更安全。这三条哲学其实是一种谦虚的工程态度承认 Agent 会犯错所以需要可观测性承认人类比 Agent 更擅长判断所以关键决策需要人工确认承认复杂度会放大错误所以追求简单六、总结这篇文章是这个系列的终章。从第 1 篇的复杂度分析到第 20 篇的工程哲学核心线索只有一条AI 工程不是消灭不确定性而是管理不确定性。Agent 的工程哲学三条简单优于复杂用最少的组件完成最多的任务可观测优于自动化先能看见再谈自动安全优于速度一次安全事故的代价大于所有性能优化的收益记住这三条你做的 Agent 系统不一定会更炫但一定会更稳。而在生产环境里稳 炫。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0730 资料来源索引并在发布前将具体来源贴到对应断言之后。
