当前位置：首页 > news >正文

【AI原生开发实战】6.2 Agent生产部署与可观测性

news 2026/4/26 1:20:32

学习目标

理解Agent与传统软件的可观测性差异
掌握Agent追踪的核心指标与实现方法
了解多Agent系统的可观测性挑战
掌握主流Agent可观测性工具的使用

一、Agent可观测性的特殊性

1.1 为什么传统监控「失灵」了

当你把精心调教的Agent部署到生产环境后，传统监控会给你展示一片绿色——QPS正常、延迟正常、错误率为零。但用户反馈却是：「AI答非所问」「任务执行到一半卡住了」。

这是因为传统监控面向的是确定性程序：

请求 → 处理 → 响应
每个步骤都是可预测的

而Agent面对的是不确定性推理：

请求 → 思考（LLM推理） → 决策 → 工具调用 → 观察 → 再思考 → … → 响应
每一步都可能有多种路径

传统APM只能看到「API调用树」，看不到Agent内部的「推理决策树」。

1.2 Agent可观测性的三层数据

要让Agent可被观测，我们需要采集三类数据：

意图与状态流：用户请求是什么？多轮对话中意图如何演变？Agent当前处于哪个状态？

完整推理轨迹：这是核心。Agent的思考过程（Chain-of-Thought）、工具选择决策、中间结果、最终判断。串联这些步骤形成「决策链」。

资源与成本维度：消耗了多少Token？调用了多少次API？每次推理的延迟分布？

# 一个典型的Agent追踪数据trace={"trace_id":"abc123","session_id":"user_session_001","agent_id":"customer_service_v2","spans":[{"span_id":"1","name":"understand_intent","start_time":"2026-04-21T02:30:00Z","duration_ms":150,"input":"用户询问订单状态","output":"用户意图: 查询订单状态, 订单ID: null"},{"span_id":"2","name":"retrieve_order_info","parent_id":"1","start_time":"2026-04-21T02:30:00.150Z","duration_ms":320,"tool":"order_api.get_order_status","input":{"user_id":"u123"},"output":{"status":"shipped","tracking":"SF123456"}},{"span_id":"3","name":"generate_response","parent_id":"1","start_time":"2026-04-21T02:30:00.470Z","duration_ms":890,"input_tokens":128,"output_tokens":86,"model":"gpt-4"}],"metrics":{"total_tokens":214,"total_cost_usd":0.00428,"total_duration_ms":1360}}

二、核心指标体系

2.1 性能指标

TTFT（Time to First Token）：从请求到生成第一个token的时间。这是用户感知延迟的关键指标。

defmeasure_ttft(agent,prompt):start=time.time()response=agent.generate(prompt,stream=True)first_token_time=Noneforchunkinresponse:iffirst_token_timeisNone:first_token_time=time.time()# 处理后续token...ttft=(first_token_time-start)*1000# 毫秒returnttft

推理延迟（Inference Latency）：纯LLM推理耗时，排除网络和工具调用开销。

端到端延迟（End-to-End Latency）：从接收请求到返回完整响应的时间。

2.2 质量指标

Agent的质量不能简单用「对/错」衡量，需要多维度评估：

任务完成率：Agent是否成功完成了用户请求？

defevaluate_task_completion(trace):""" 判断任务是否完成 逻辑： 1. 如果所有必要工具都被调用且成功 → 完成 2. 如果调用了错误工具或工具失败 → 未完成 3. 如果提前终止 → 部分完成 """required_tools=trace.metadata.required_tools called_tools=[span.toolforspanintrace.spansifspan.type=="tool"]ifall(toolincalled_toolsfortoolinrequired_tools):return"completed"elifany(toolincalled_toolsfortoolinrequired_tools):return"partial"else:return"failed"

意图识别准确率：Agent是否正确理解了用户意图？

响应质量评分：使用LLM-as-Judge评估回复质量。

2.3 成本指标

Token消耗直接关联运营成本：

classCostTracker:def__init__(self):self.total_input_tokens=0self.total_output_tokens=0self.total_cost=0.0# 各模型定价（$/1000 tokens）self.pricing={"gpt-4":{"input":0.03,"output":0.06},"gpt-3.5-turbo":{"input":0.0015,"output":0.002},"claude-3":{"input":0.015,"output":0.075}}defrecord_usage(self,model,input_tokens,output_tokens):self.total_input_tokens+=input_tokens self.total_output_tokens+=output_tokens price=self.pricing.get(model,{"input":0,"output":0})cost=(input_tokens/1000)*price["input"]+\(output_tokens/1000)*price["output"]self.total_cost+=costdefget_report(self):return{"total_input_tokens":self.total_input_tokens,"total_output_tokens":self.total_output_tokens,"total_cost_usd":round(self.total_cost,4)}

三、追踪实现

3.1 OpenTelemetry集成

OpenTelemetry是云原生可观测性的标准，让Agent追踪变得标准化：

fromopentelemetryimporttracefromopentelemetry.sdk.traceimportTracerProviderfromopentelemetry.sdk.trace.exportimportBatchSpanProcessorfromopentelemetry.sdk.resourcesimportResourcefromopentelemetry.semconv.resourceimportResourceAttributes# 初始化追踪器resource=Resource.create({ResourceAttributes.SERVICE_NAME:"customer-service-agent",ResourceAttributes.SERVICE_VERSION:"2.0.0"})trace.set_tracer_provider(TracerProvider(resource=resource))tracer=trace.get_tracer(__name__)# 在Agent代码中使用classTracedAgent:def__init__(self,agent):self.agent=agentasyncdefrun(self,prompt):withtracer.start_as_current_span("agent_run")asspan:span.set_attribute("user.prompt",prompt[:100])# 截断避免过大# 追踪LLM调用withtracer.start_as_current_span("llm_inference")asllm_span:response=awaitself.agent.llm.generate(prompt)llm_span.set_attribute("llm.model",self.agent.llm.model)llm_span.set_attribute("llm.input_tokens",response.usage.input_tokens)llm_span.set_attribute("llm.output_tokens",response.usage.output_tokens)# 追踪工具调用fortool_callinresponse.tool_calls:withtracer.start_as_current_span(f"tool.{tool_call.name}")astool_span:tool_span.set_attribute("tool.name",tool_call.name)tool_span.set_attribute("tool.args",str(tool_call.args))result=awaitself.execute_tool(tool_call)tool_span.set_attribute("tool.result",str(result)[:200])returnresponse

3.2 追踪数据模型

Agent追踪需要结构化的数据模型来表达推理过程：

fromdataclassesimportdataclass,fieldfromtypingimportList,Dict,Optional,AnyfromenumimportEnumimporttimeclassSpanType(Enum):LLM_INFERENCE="llm_inference"TOOL_CALL="tool_call"TOOL_RESULT="tool_result"REASONING="reasoning"USER_INPUT="user_input"AGENT_RESPONSE="agent_response"@dataclassclassAgentSpan:span_id:strparent_id:Optional[str]name:strspan_type:SpanType start_time:floatend_time:float=0# 类型特定属性llm_info:Optional[Dict]=None# LLM调用信息tool_info:Optional[Dict]=None# 工具调用信息reasoning_info:Optional[Dict]=None# 推理步骤信息@propertydefduration_ms(self)->float:return(self.end_time-self.start_time)*1000@dataclassclassAgentTrace:trace_id:strsession_id:stragent_id:struser_id:strcreated_at:float=field(default_factory=time.time)spans:List[AgentSpan]=field(default_factory=list)metadata:Dict[str,Any]=field(default_factory=dict)# 汇总指标total_duration_ms:float=0total_tokens:int=0total_cost:float=0tool_call_count:int=0

四、Agent评估体系

4.1 Code-as-Judge

对于有明确规则的评估，使用代码验证：

defevaluate_code_agent(trace)->EvaluationResult:""" 代码生成Agent评估 检查点： 1. 代码是否生成 2. 代码语法是否正确 3. 代码是否产生预期输出 """result=EvaluationResult()# 检查点1：代码生成code_spans=[sforsintrace.spansifs.name=="code_generation"]ifnotcode_spans:result.add_error("No code was generated")returnresult generated_code=code_spans[0].tool_info.get("code","")result.metrics["code_length"]=len(generated_code)# 检查点2：语法正确性try:compile(generated_code,'<string>','exec')result.metrics["syntax_valid"]=TrueexceptSyntaxErrorase:result.add_error(f"Syntax error:{e}")result.metrics["syntax_valid"]=False# 检查点3：执行结果ifresult.metrics.get("syntax_valid"):exec_result=execute_sandbox(generated_code,timeout=5)ifexec_result.success:result.metrics["execution_success"]=Trueresult.metrics["output"]=exec_result.outputelse:result.add_error(f"Execution failed:{exec_result.error}")result.metrics["execution_success"]=Falsereturnresult

4.2 LLM-as-Judge

对于开放性评估，使用LLM裁判：

classLLMJudge:def__init__(self,judge_model="gpt-4"):self.llm=LLM(judge_model)self.judge_prompt_template=""" You are an expert evaluator judging an AI Agent's response. User Request: {user_request} Agent Response: {agent_response} Evaluation Criteria: 1. Relevance: Does the response directly address the user's request? 2. Helpfulness: Is the response useful and informative? 3. Safety: Does the response avoid harmful content? 4. Coherence: Is the response logically coherent? Provide your evaluation in JSON format: {{ "relevance_score": 1-5, "helpfulness_score": 1-5, "safety_score": 1-5, "coherence_score": 1-5, "overall_score": 1-5, "feedback": "Brief explanation of your evaluation" }} """defevaluate(self,user_request:str,agent_response:str)->Dict:prompt=self.judge_prompt_template.format(user_request=user_request,agent_response=agent_response)response=self.llm.generate(prompt)returnjson.loads(response.content)

4.3 评估飞轮

评估不是为了打分，而是形成「评估-分析-优化」的闭环：

classEvaluationFlywheel:def__init__(self,agent,evaluator,db):self.agent=agent self.evaluator=evaluator self.db=dbdefrun_evaluation_cycle(self,test_cases:List[TestCase]):results=[]fortest_caseintest_cases:# 执行Agenttrace=self.agent.run(test_case.prompt)# 评估evaluation=self.evaluator.evaluate(trace)# 存储结果self.db.save({"test_case_id":test_case.id,"trace":trace,"evaluation":evaluation,"timestamp":datetime.now()})results.append(evaluation)# 分析结果，生成改进建议analysis=self.analyze_results(results)returnEvaluationReport(results=results,analysis=analysis)defanalyze_results(self,results):# 识别失败模式failed_cases=[rforrinresultsifr.overall_score<3]# 提取常见错误common_errors=self.extract_common_patterns(failed_cases)# 生成改进建议suggestions=[]forerrorincommon_errors:suggestions.append(self.generate_improvement_suggestion(error))return{"success_rate":sum(1forrinresultsifr.overall_score>=3)/len(results),"average_score":sum(r.overall_scoreforrinresults)/len(results),"common_errors":common_errors,"improvement_suggestions":suggestions}

五、主流工具对比

5.1 Langfuse

Langfuse是专门为LLM应用设计的可观测性平台：

# Langfuse集成fromlangfuseimportLangfuse langfuse=Langfuse()classLangfuseTrackedAgent:def__init__(self,agent):self.agent=agent self.langfuse=langfusedefrun(self,prompt,user_id):# 创建追踪trace=self.langfuse.trace(name="agent_run",user_id=user_id,metadata={"agent_version":"2.0"})# 使用上下文管理器自动追踪withtrace.span("llm_call")asspan:response=self.agent.llm.generate(prompt)span.input=prompt span.output=response.content span.metrics={"input_tokens":response.usage.input_tokens,"output_tokens":response.usage.output_tokens,"latency":response.latency}returnresponse

5.2 MLflow

MLflow的Tracking模块支持Agent追踪：

importmlflow# 配置MLflow追踪mlflow.set_tracking_uri("http://mlflow-server:5000")mlflow.langchain.autolog()# Agent执行自动被追踪withmlflow.start_span("agent_execution")asspan:response=agent.run(prompt)span.set_attribute("output_length",len(response.content))span.set_attribute("tool_calls",len(response.tool_calls))

5.3 工具选型决策

工具	优势	劣势	推荐场景
Langfuse	专为LLM设计，开箱即用	托管成本	快速搭建可观测性
MLflow	生态完整，实验管理强	配置复杂	已有MLflow基础设施
Phoenix	免费开源，本地部署	功能相对基础	自托管需求
LangSmith	与LangChain深度集成	绑定LangChain	使用LangChain开发

六、实战：构建可观测Agent系统

6.1 架构设计

┌─────────────────────────────────────────────────────────┐ │ Agent Service │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Agent │ │ Tracer │ │ Logger │ │ │ │ Logic │──│ (OTel) │──│ (结构化) │ │ │ └─────────────┘ └──────┬──────┘ └──────┬──────┘ │ └──────────────────────────┼──────────────────┼──────────┘ │ │ ┌────────────────┘ │ ▼ ▼ ┌─────────────────┐ ┌─────────────────────┐ │ OTEL Collector │ │ Elasticsearch │ │ (Trace聚合) │ │ (日志存储) │ └────────┬────────┘ └──────────┬──────────┘ │ │ ▼ ▼ ┌─────────────────┐ ┌─────────────────────┐ │ Jaeger/Grafana │ │ Kibana │ │ (追踪可视化) │ │ (日志分析) │ └─────────────────┘ └─────────────────────┘ │ ▼ ┌─────────────────┐ │ Prometheus │ │ (指标聚合) │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ Grafana │ │ (仪表盘) │ └─────────────────┘

6.2 关键实现点

追踪数据采样：不是所有请求都需要完整追踪。

classSmartSampler:def__init__(self,sample_rate=0.1):self.sample_rate=sample_rate self.priority_rules=[("error",1.0),# 错误请求100%采样("slow",1.0),# 慢请求100%采样("user_id",0.8),# VIP用户高采样("default",sample_rate)# 普通请求按比例采样]defshould_sample(self,request,response):# 高优先级：错误或慢请求ifresponse.is_errororresponse.duration_ms>5000:returnTrue# VIP用户ifrequest.user_idinVIP_USERS:returnTrue# 随机采样returnrandom.random()<self.sample_rate

敏感数据脱敏：追踪数据可能包含用户隐私。

classDataSanitizer:PII_PATTERNS=[(r'\b\d{11}\b','[PHONE]'),# 手机号(r'\b\d{18}\b','[ID]'),# 身份证(r'\b[\w.-]+@[\w.-]+\.\w+\b','[EMAIL]'),# 邮箱]defsanitize(self,data):ifisinstance(data,str):forpattern,replacementinself.PII_PATTERNS:data=re.sub(pattern,replacement,data)elifisinstance(data,dict):return{k:self.sanitize(v)fork,vindata.items()}elifisinstance(data,list):return[self.sanitize(item)foritemindata]returndata