企业官网建设流程全解析

[Agent Run] 总耗时: 12.3s, tokens: 4,280 ├── [LLM Call #1] 意图解析 │ ├── input_tokens: 520, output_tokens: 85 │ ├── duration: 1.2s │ └── model: gpt-6.1-sol ├── [Tool Call] database_query │ ├── status: success │ ├── duration: 0.3s │ └── result_rows: 15 ├── [LLM Call #2] 结果推理 │ ├── input_tokens: 1,240, output_tokens: 320 │ ├── duration: 2.1s │ └── model: gpt-6.1-sol ├── [Tool Call] send_email │ ├── status: success │ └── duration: 0.8s └── [LLM Call #3] 最终回复生成 ├── input_tokens: 1,800, output_tokens: 315 ├── duration: 1.8s └── model: gpt-6.1-sol

pip install opentelemetry-api opentelemetry-sdk \ opentelemetry-exporter-otlp-proto-http \ openai arize-phoenix

pip install arize-phoenix python -m phoenix.server.main serve # 默认监听 http://localhost:6006

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource

定义资源——告诉后端"我是谁"

resource = Resource.create({
"service.name": "my-ai-agent",
"service.version": "1.0.0",
"deployment.environment": "production"
})

provider = TracerProvider(resource=resource)

导出到Phoenix（本地开发）

生产环境换成你的OTLP Collector地址

exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
provider.add_span_processor(BatchSpanProcessor(exporter))

trace.set_tracer_provider(provider)
tracer = trace.get_tracer(name)

第二步：封装LLM调用的追踪装饰器

这是整个方案的核心——每个LLM调用自动生成一个带有标准属性的Span。

import time import json import functools from opentelemetry import trace from openai import OpenAI

client = OpenAI()

def trace_llm_call(model=None):
“”“装饰器：自动追踪LLM调用的token消耗、耗时和输入输出”“”
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
tracer = trace.get_tracer(name)
span_name = f"llm_call:{model or ‘unknown’}"
with tracer.start_as_current_span(span_name) as span:
start_time = time.time()

# 设置模型信息 span.set_attribute("gen_ai.system", "openai") span.set_attribute("gen_ai.request.model", model or "gpt-4-turbo") # 记录输入（生产环境注意脱敏） if args and isinstance(args[0], list): messages = args[0] span.set_attribute("gen_ai.prompt", json.dumps(messages[-1], ensure_ascii=False)[:500]) try: response = func(*args, **kwargs) # 记录token消耗 usage = response.usage span.set_attribute("gen_ai.usage.prompt_tokens", usage.prompt_tokens) span.set_attribute("gen_ai.usage.completion_tokens", usage.completion_tokens) # 记录耗时 duration = time.time() - start_time span.set_attribute("gen_ai.response.duration_ms", int(duration * 1000)) # 记录输出摘要（截断防止trace过大） content = response.choices[0].message.content span.set_attribute("gen_ai.completion", content[:300] if content else "") return response except Exception as e: span.set_status(trace.StatusCode.ERROR) span.set_attribute("gen_ai.error.message", str(e)) raise return wrapper return decorator

使用方式非常直观：

@trace_llm_call(model="gpt-6.1-sol") def call_agent_brain(messages): """Agent的核心推理函数""" return client.chat.completions.create( model="gpt-6.1-sol", messages=messages, temperature=0.3, max_tokens=2000 )

第三步：封装工具调用的追踪

工具调用的追踪逻辑不同——我们要记录的是工具名、参数、执行结果和状态。

def trace_tool_call(tool_name): """装饰器：追踪工具调用的参数、结果和执行状态""" def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): tracer = trace.get_tracer(__name__) span_name = f"tool_call:{tool_name}" with tracer.start_as_current_span(span_name) as span: start_time = time.time()
span.set_attribute("gen_ai.tool.name", tool_name) # 记录调用参数（脱敏处理） safe_kwargs = { k: str(v)[:200] for k, v in kwargs.items() } span.set_attribute("gen_ai.tool.input", json.dumps(safe_kwargs, ensure_ascii=False)) try: result = func(*args, **kwargs) duration = time.time() - start_time span.set_attribute("gen_ai.tool.status", "success") span.set_attribute("gen_ai.tool.duration_ms", int(duration * 1000)) # 记录结果摘要 span.set_attribute("gen_ai.tool.output", str(result)[:300]) return result except Exception as e: span.set_attribute("gen_ai.tool.status", "error") span.set_attribute("gen_ai.tool.error", str(e)) span.set_status(trace.StatusCode.ERROR) raise return wrapper return decorator

使用示例

@trace_tool_call("database_query")
def query_database(sql, params=None):
"""执行数据库查询"""
# 实际的数据库查询逻辑
return {"rows": 15, "data": []}

第四步：串联Agent运行——追踪整条链路

现在把LLM调用和工具调用串起来，追踪一次完整的Agent运行。

import uuid

def run_agent(user_input):
"""运行Agent并追踪完整执行链"""
agent_id = str(uuid.uuid4())

# 创建Agent级别的根Span with tracer.start_as_current_span("agent_run") as root_span: root_span.set_attribute("gen_ai.agent.id", agent_id) root_span.set_attribute("gen_ai.agent.name", "customer_service_agent") root_span.set_attribute("gen_ai.agent.turn_index", 0) messages = [ {"role": "system", "content": "你是一个智能客服助手..."}, {"role": "user", "content": user_input} ] # 第一轮：意图解析 response = call_agent_brain(messages) content = response.choices[0].message.content # 检查是否需要调用工具 if "query_order" in content: order_id = extract_order_id(content) order_info = query_database( sql="SELECT * FROM orders WHERE id = %s", params={"id": order_id} ) # 第二轮：基于工具结果生成回复 messages.append({"role": "assistant", "content": content}) messages.append({"role": "tool", "content": json.dumps(order_info)}) final_response = call_agent_brain(messages) return final_response.choices[0].message.content return content

这样，当Agent运行时，Phoenix（或者你选择的后端）里就能看到一棵完整的调用树：

agent_run (gen_ai.agent.id = xxx, turn_index = 0) ├── llm_call:gpt-6.1-sol (prompt_tokens=520, completion_tokens=85, 1200ms) ├── tool_call:database_query (status=success, 300ms) └── llm_call:gpt-6.1-sol (prompt_tokens=1240, completion_tokens=320, 2100ms)

第五步：基于Trace数据做异常检测

有了完整的trace数据，接下来就可以做有意思的事了——用规则引擎检测异常行为模式。

from collections import defaultdict from datetime import datetime, timedelta

class AgentBehaviorMonitor:
"""基于trace数据的Agent行为监控"""

def __init__(self): self.tool_call_history = defaultdict(list) self.loop_threshold = 3 # 连续相同调用超过3次触发告警 def check_tool_call(self, agent_id, tool_name, arguments, timestamp): """检查工具调用是否异常""" history = self.tool_call_history[agent_id] history.append({ "tool": tool_name, "args_hash": hash(json.dumps(arguments, sort_keys=True)), "time": timestamp }) # 检测1：循环调用检测 recent = history[-self.loop_threshold:] if len(recent) >= self.loop_threshold: all_same_tool = all(c["tool"] == tool_name for c in recent) all_same_args = all( c["args_hash"] == recent[0]["args_hash"] for c in recent ) if all_same_tool and all_same_args: return { "alert": "LOOP_DETECTED", "detail": f"Agent {agent_id} 连续{self.loop_threshold}" f"次调用 {tool_name} 且参数相同", "action": "CIRCUIT_BREAK" } # 检测2：高频调用检测 one_minute_ago = timestamp - timedelta(minutes=1) calls_last_minute = [ c for c in history if c["time"] > one_minute_ago ] if len(calls_last_minute) > 20: return { "alert": "HIGH_FREQUENCY", "detail": f"Agent {agent_id} 1分钟内调用了" f"{len(calls_last_minute)}次工具", "action": "RATE_LIMIT" } # 检测3：敏感操作检测 sensitive_tools = ["database_write", "send_email", "delete_resource"] if tool_name in sensitive_tools: return { "alert": "SENSITIVE_OPERATION", "detail": f"Agent执行了敏感操作: {tool_name}", "action": "REQUIRE_HUMAN_APPROVAL" } return None

这段代码看着简单，但在生产环境中能帮你挡住大部分灾难。

比如Agent陷入死循环（反复用同样的参数调用同一个工具），传统做法是等它把额度烧完你才发现。有了行为监控，3次相同调用就触发熔断。

再比如Agent突然开始调用敏感工具（写数据库、发邮件），系统自动要求人工确认后才执行。

第六步：采样策略——别让trace数据把你的存储吃垮

生产环境有个现实问题：每条trace都完整记录，数据量会非常大。LLM调用的Span包含prompt和response文本，单条trace可能就有几十KB。

推荐的采样策略（来源：Uptrace 2026年OTel最佳实践文档）：

import random

class SmartSampler:
"""智能采样：错误和慢请求全量记录，正常请求按比例采样"""

def __init__(self, normal_sample_rate=0.1): self.normal_rate = normal_sample_rate def should_sample(self, span) -> bool: # 错误的Span全量记录 if span.status.status_code == trace.StatusCode.ERROR: return True # 耗时超过30秒的Span全量记录 duration_ms = span.attributes.get("gen_ai.response.duration_ms", 0) if duration_ms > 30000: return True # token消耗超过5000的Span全量记录 total_tokens = ( span.attributes.get("gen_ai.usage.prompt_tokens", 0) + span.attributes.get("gen_ai.usage.completion_tokens", 0) ) if total_tokens > 5000: return True # 正常请求按比例采样 return random.random() < self.normal_rate

这个策略的核心逻辑：正常的调用采样10%就够了（省存储），但出错、慢、贵的调用必须100%记录——因为这些恰恰是你需要排查的。

监控平台选型：5个工具实测对比

后端存储和可视化这块，2026年主流的选择大概有这几个（以下数据来源于befailproof.ai 2026年10月的横向评测）：

我的建议：

• 预算紧张 + 想自托管：选Langfuse。MIT协议，5人团队100万trace/月大约$69（vs LangSmith约$420）
• 全栈LangChain：选LangSmith，零配置省心
• 想自己控制一切：选Phoenix + ClickHouse，完全开源
• 已有Datadog：直接用Datadog的LLM Observability，省一套系统

不管选哪个，前端接入都是OTLP协议——上面写的代码一行都不用改。这也是OpenTelemetry的价值所在。

几个踩过的坑

最后分享几个我在实际项目中踩过的坑，帮你省点时间。

坑1：Trace数据里包含了用户敏感信息。

Prompt里可能有用户的手机号、身份证号。如果你直接把prompt和response存进trace，这些数据就进入了可观测性系统。解决方案是在记录前做一层脱敏：

import re

def sanitize_text(text):
“”“脱敏处理：替换手机号、身份证等敏感信息”“”
text = re.sub(r’1[3-9]\d{9}‘, ‘PHONE’, text)
text = re.sub(r’\d{17}[\dXx]', ‘ID_CARD’, text)
return text

坑2：Agent的多轮对话导致trace树太深。

如果一个Agent有10轮工具调用，每轮都有LLM调用，trace树会变得非常深。建议设置最大追踪深度，超过深度的调用合并为一个聚合Span。

坑3：别忘了给trace加上业务标签。

纯技术属性（token数、耗时）不够用。你需要加上业务维度的标签——用户ID、租户ID、业务场景、对话ID。这样你才能回答"某个租户的Agent平均消耗是不是异常偏高"这种问题。

span.set_attribute("business.tenant_id", tenant_id) span.set_attribute("business.scenario", "customer_service") span.set_attribute("business.conversation_id", conv_id)

总结

AI Agent和传统微服务最大的区别在于：传统服务是确定性的（同样的输入同样的输出），Agent是概率性的（同样的输入，行为可能完全不同）。

这意味着传统APM那一套——看延迟、看错误率、看CPU——对Agent基本不够用。你需要知道它选择了哪个工具、为什么这么选、选完之后结果如何、有没有跑偏。

OpenTelemetry + GenAI语义规范 + 行为监控引擎，是目前我能想到的最务实的方案。不是最炫酷的，但确实是最管用的。

如果你们团队也在做Agent的可观测性，欢迎在评论区聊聊你们用了什么方案。踩过什么坑也说说，让大家少走弯路。

本文代码基于OpenTelemetry Python SDK 1.28+，Phoenix 8.x测试通过。文中引用的安全事件数据来源于OpenAI和Anthropic 2026年7月公开披露的安全报告。