七月 Agent 实践终章：从 Demo 到产品还差一套工程体系

七月 Agent 实践终章从 Demo 到产品还差一套工程体系一、个性化深度引言七月初我用LangGraph搭建了一个Agent Demo——能自动拆解任务、调用5个工具、生成分析报告。Demo演示时20次测试成功了18次失败率10%。看起来不错。然后我把这个Agent接入了真实的业务数据流。在真实环境中失败率从10%飙升到了47%。问题不在于Agent不够聪明而在于真实环境有太多Demo中不存在的情况API超时、工具返回异常格式、上下文窗口溢出、用户中途改变需求……见证奇迹的时刻不在Demo演示室里而在生产环境的监控面板上——当失败率从47%逐步降到8%时降低的不是智商而是系统工程能力。本文总结七月Agent实践最深刻的教训Demo和产品之间隔着的不是算法是工程。二、个性化原理剖析Agent从Demo到产品的工程化差距异常处理是最大的差距。Demo假设一切正常——API返回标准JSON工具调用总是成功用户输入总是完整。真实环境中这些假设都不成立。七月的数据47%的失败中35%来自API超时或异常返回8%来自格式解析失败只有4%是推理逻辑错误。状态管理决定系统上限。Demo中的任务通常在3-5步内完成上下文窗口足够。但真实的多步骤Agent任务可能需要20步。如果不做状态管理优化上下文压缩、中间结果存储系统在第15步左右就会因为上下文溢出而失败。可观测性是调试的基础。Demo出问题时可以直接看日志。但在生产环境中Agent可能同时在处理数十个任务传统日志已经无法满足调试需求。需要链路追踪每个任务的全生命周期视图、结构化日志可搜索、可聚合、实时监控异常告警。三、个性化代码实践产品级Agent的可观测性和容错框架import uuid import time import traceback from typing import Dict, List, Any, Optional, Callable from dataclasses import dataclass, field from datetime import datetime from enum import Enum class StepStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed RETRY retry TIMEOUT timeout dataclass class TraceSpan: 单步追踪 span_id: str step_name: str start_time: float # 设计原因input/output记录每一步的完整数据 # 支持事后完整回放和问题复现 input_data: Dict[str, Any] output_data: Optional[Dict[str, Any]] None status: StepStatus StepStatus.PENDING # 设计原因error_details存储异常堆栈和上下文 # 结构化错误信息比原始traceback更易于搜索和聚合 error_details: Optional[Dict[str, Any]] None duration_ms: float 0 retry_count: int 0 dataclass class TaskTrace: 任务全链路追踪 task_id: str task_name: str # 设计原因user_input保留原始用户输入 # 支持问题复现——用同样的输入重新执行 user_input: str spans: List[TraceSpan] field(default_factorylist) started_at: float field(default_factorytime.time) completed_at: Optional[float] None final_status: StepStatus StepStatus.PENDING class ProductionAgentRunner: 产品级Agent运行器 设计原因核心变化——从跑通任务到可靠地跑通任务。 每个步骤都加了追踪、重试、超时保护。 这些不是功能代码是可靠性代码。 def __init__(self, max_retries: int 3, step_timeout: float 30.0): self.max_retries max_retries self.step_timeout step_timeout self._traces: Dict[str, TaskTrace] {} def start_task(self, task_name: str, user_input: str) - str: 开始追踪一个任务 task_id str(uuid.uuid4())[:12] self._traces[task_id] TaskTrace( task_idtask_id, task_nametask_name, user_inputuser_input ) return task_id def execute_step( self, task_id: str, step_name: str, step_fn: Callable, step_input: Dict[str, Any] ) - Dict[str, Any]: 执行单个步骤带追踪和重试 设计原因每个步骤都被追踪和记录 不是执行完就忘的Demo模式 而是事事留痕的产品模式。 trace self._traces.get(task_id) if not trace: raise ValueError(f任务 {task_id} 不存在) span TraceSpan( span_idstr(uuid.uuid4())[:8], step_namestep_name, start_timetime.time(), input_datastep_input ) trace.spans.append(span) # 设计原因重试机制放在步骤级别而非任务级别 # 因为不同步骤的失败模式不同需要独立重试策略 for attempt in range(self.max_retries 1): try: import asyncio if asyncio.iscoroutinefunction(step_fn): # 异步函数特殊处理 pass span.status StepStatus.RUNNING start time.time() output step_fn(step_input) span.duration_ms (time.time() - start) * 1000 span.output_data output span.status StepStatus.SUCCESS span.retry_count attempt return output except Exception as e: span.retry_count attempt if attempt self.max_retries: span.status StepStatus.RETRY # 设计原因指数退避防止重试风暴 # 1s→2s→4s的间隔比等间隔更合理 import time time.sleep(2 ** attempt) else: span.status StepStatus.FAILED span.error_details { error_type: type(e).__name__, error_message: str(e), traceback: traceback.format_exc(), attempt: attempt 1 } # 设计原因失败时返回结构化错误而非抛异常 # 让上层能根据错误类型选择不同的处理策略 return { success: False, error: str(e), step: step_name, retry_count: attempt } return {success: False, error: max_retries_exceeded} def get_task_report(self, task_id: str) - Dict[str, Any]: 获取任务执行报告 trace self._traces.get(task_id) if not trace: return {} total_duration ( (trace.completed_at or time.time()) - trace.started_at ) * 1000 failed_steps [ s for s in trace.spans if s.status StepStatus.FAILED ] retried_steps [ s for s in trace.spans if s.retry_count 0 ] return { task_id: task_id, task_name: trace.task_name, total_steps: len(trace.spans), failed_steps: len(failed_steps), retried_steps: len(retried_steps), total_duration_ms: total_duration, success_rate: ( 1 - len(failed_steps) / max(len(trace.spans), 1) ), bottleneck_step: ( max(trace.spans, keylambda s: s.duration_ms).step_name if trace.spans else None ), spans: [ { step: s.step_name, status: s.status.value, duration_ms: s.duration_ms, retries: s.retry_count } for s in trace.spans ] }四、个性化边界权衡可靠性代码 vs 功能代码。可靠性代码重试、超时、降级、日志通常占产品代码的40-50%。Demo中这些代码为0。增加的开发成本换来的是任务成功率从60%到95%的提升。在预算有限时优先保证核心路径的可靠性非核心路径用简单策略。实时重试 vs 异步补偿。实时重试简单直接但会阻塞任务管道。异步补偿先返回部分结果后续修正不阻塞管道但实现复杂。高频、快速的API调用适合实时重试。低频、慢速的工具调用适合异步补偿。全量追踪 vs 采样追踪。全量追踪保留所有信息但存储和查询成本高。采样追踪成本低但可能丢失关键错误。推荐方案正常case做采样10%错误case全量追踪。单Agent vs 多Agent。单Agent架构简单、可观测性好但难以处理复杂任务。多Agent架构能力强但可观测性大幅下降——一个问题可能是由多个Agent的交互造成的难以定位根因。七月选择先做好单Agent的可靠性再考虑多Agent的扩展性。五、总结Agent从Demo到产品的差距本质上是工程体系的差距。Demo关心能不能跑通产品关心能不能一直在跑而且不出问题。填补这个差距需要六个方面的工程化异常处理、状态管理、可观测性、容错与重试、资源管理、版本管理。其中可观测性是最容易被忽视但最重要的——没有足够的观测数据你永远不知道系统在哪里出问题只能靠猜。八月Agent工作的重点不是增加新功能而是把现有功能的可靠性从60%提升到95%。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0731 资料来源索引并在发布前将具体来源贴到对应断言之后。