☰
从可观测到可优化:用 AgentInsight SDK 打造 AI Agent 全链路监控与持续调优体系
2026/10/8 12:29:24 网站建设 项目流程

1. Agent 上线后为什么总在“盲调”:从一次线上延迟抖动说起

AI Agent 从 Demo 走到生产,最折磨人的不是模型能力不够,而是“看不见、管不住、改不了”。我试过最典型的一次:用户反馈“回答变慢了”,我打开日志只看到一行POST /api/chat 200 OK,至于中间模型调了几次、检索命中了哪些文档、工具返回了什么、Token 花了多少,全靠猜。这不是个例,而是 Agent 应用在生产环境里的系统性工程缺失。

传统 Web 应用有 APM 覆盖接口耗时、错误率、资源使用,但 Agent 的核心问题完全不同。它的执行链路是动态的:规划 → 检索 → 推理 → 工具 → 校验 → 生成,每一步都可能分叉。失败模式也不是 HTTP 状态码能表达的,而是幻觉、格式错误、工具选错、上下文漂移。成本结构更麻烦,传统服务是服务器固定成本,Agent 是 Token 动态成本,随链路复杂度非线性增长。质量评估单元测试也测不出来,需要持续评估输出质量、工具准确率、检索相关性。

AgentInsight SDK 就是为这类问题设计的开源可观测工具,基于 OpenTelemetry 协议,支持 Python 和 TypeScript,可以零侵入接入现有 Agent 应用,采集 Trace 链路、模型调用、Token 消耗、响应耗时、异常错误等关键数据。它解决的是“可观测 → 可监控 → 可优化”三层闭环,适合已经上线或即将上线 AI Agent、需要定位延迟与失败环节的工程团队。

这篇内容不重复讨论“为什么 Agent 需要可观测”,而是直接进入工程实践:怎么用 AgentInsight SDK 一步步搭建可观测、可监控、可优化的 Agent 工程体系,同时结合 TaoToken 统一 Key/API 通道打通调用日志与指标,让链路数据真正可用。

2. 前置准备:TaoToken 统一通道与 AgentInsight SDK 安装配置

在接入 AgentInsight 之前,先把模型调用通道统一起来。很多团队的问题是:Agent 里同时调了 OpenAI、Claude、国产模型,每个 Key 分散在不同环境变量里,日志也对不上。TaoToken 提供统一的 API 通道,把模型调用收敛到一个 Base URL 和一把 Key,这样 AgentInsight 采集到的 Trace 才能和调用日志一一对应。

TaoToken 的 API 地址是https://taotoken.net/api,官网入口是https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=。你可以在控制台创建 API Key,然后在 Agent 项目里统一配置。这样做的好处是:无论底层换哪个模型,AgentInsight 里的 Trace 结构不变,调优时对比的是同一套指标。

先安装 AgentInsight SDK。Python 版本要求 3.10 以上:

pip install agentinsight-sdk pip install agentinsight-sdk openai pip install agentinsight-sdk langchain langchain-openai

初始化有两种方式。生产环境推荐环境变量:

export AGENTINSIGHT_PUBLIC_KEY="pk-..." export AGENTINSIGHT_SECRET_KEY="sk-..." export AGENTINSIGHT_BASE_URL="https://agent.goldebridge.com" export TAOTOKEN_API_KEY="你的 TaoToken Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

代码里直接初始化:

from agentinsight import AgentInsight client = AgentInsight( public_key="pk-...", secret_key="sk-...", base_url="https://agent.goldebridge.com", tracing_enabled=True, flush_at=512, flush_interval=5.0, sample_rate=1.0, environment="production", release="1.0.0", )

这里几个参数值得说明。flush_at=512是批量发送阈值,攒够 512 条 Span 才发一次,避免高频小包拖慢业务。flush_interval=5.0是兜底间隔,即使没攒够也会 5 秒发一次。sample_rate=1.0是全量采集,生产环境流量大时可以降到 0.1~0.3。environment和release是标签,方便在平台上按版本过滤 Trace。

如果你用 TaoToken 统一通道,模型客户端这样配:

from openai import OpenAI llm_client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], )

这样 AgentInsight 采集到的 generation Span 里,模型调用地址就是 TaoToken 通道,日志和指标能对齐。踩过的坑是:有人把base_url写成带/v1的路径,结果 SDK 拼接后变成/v1/v1/chat/completions,报 404。TaoToken 的 Base URL 就是https://taotoken.net/api,不要自己加后缀。

3. 可复制配置:@observe 埋点字段清单与 JSON/TOML 片段

AgentInsight 最核心的能力是@observe装饰器,它自动追踪函数调用,嵌套时自动建立父子关系。先看一个真实 RAG Agent 场景:

from agentinsight import observe @observe(as_type="agent", name="rag-agent") def run_rag_agent(query: str) -> str: docs = retrieve_documents(query) answer = generate_answer(query, docs) validated = safety_check(answer) return validated @observe(as_type="retriever", name="document-retrieval") def retrieve_documents(query: str) -> list[str]: results = vector_store.similarity_search(query, k=3) return [doc.page_content for doc in results] @observe(as_type="generation", name="answer-generation") def generate_answer(query: str, docs: list[str]) -> str: context = "\n".join(docs) prompt = f"基于以下上下文回答问题:\n\n{context}\n\n问题:{query}" response = llm_client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.7, ) return response.choices[0].message.content @observe(as_type="guardrail", name="safety-validation") def safety_check(answer: str) -> str: if contains_sensitive_info(answer): return "[已过滤:回答包含敏感信息]" return answer

每个@observe装饰的函数自动成为一个 Span,嵌套调用时自动建立父子关系。在 AgentInsight 平台上你会看到完整 Trace:

rag-agent (agent) ├── document-retrieval (retriever) ├── answer-generation (generation) └── safety-validation (guardrail)

每个 Span 自动记录:函数名、输入/输出、开始/结束时间、执行耗时、异常信息。观察类型(as_type)的语义对照如下:

as_type 含义典型场景
agentAgent 执行主体、主循环、多步推理
chain链式调用、Prompt 模板链、预处理链
tool工具调用、外部 API、数据库查询、代码执行
generationLLM 生成、模型推理调用
retriever检索器、向量检索、全文搜索
embedding向量嵌入、文本向量化
guardrail安全护栏、输入/输出校验、内容过滤
evaluator评估器、自动评分、质量检测

正确使用这些类型,Trace 在平台上会呈现更清晰的语义结构。对于需要精细控制的场景,比如异步调用、动态 Span 创建,可以用底层 API:

from agentinsight import AgentInsight client = AgentInsight() with client.start_as_current_observation( name="process-user-query", as_type="span", ) as root_span: with root_span.start_as_current_observation( name="llm-inference", as_type="generation", model="gpt-4", input={"query": "用户的具体问题"}, model_parameters={"temperature": 0.7, "max_tokens": 500}, ) as gen_span: response = call_llm("用户的具体问题") gen_span.update( output=response, usage_details={ "prompt_tokens": 150, "completion_tokens": 300, }, cost_details={"total_cost": 0.0045}, ) client.flush()

如果你用 LangChain,接入更简单:

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from agentinsight.langchain import CallbackHandler handler = CallbackHandler() llm = ChatOpenAI( model="gpt-4", temperature=0, openai_api_key=os.environ["TAOTOKEN_API_KEY"], openai_api_base=os.environ["TAOTOKEN_BASE_URL"], ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的技术助手,请用中文回答。"), ("human", "{input}"), ]) chain = prompt | llm | StrOutputParser() result = chain.invoke( {"input": "什么是 Agent 可观测?"}, config={"callbacks": [handler]}, )

CallbackHandler 会自动追踪 LangChain 链路中的每一个环节:Prompt 模板渲染、LLM 调用、Output Parser 处理等。这里三件套必须写全:Base URL 用https://taotoken.net/api,Key 用 TaoToken 控制台创建的 Key,Model ID 按你实际调用的模型填,比如gpt-4、claude-3-5-sonnet等。

生产环境建议把配置写成 JSON 或 TOML,避免硬编码。比如agentinsight.toml:

[agentinsight] public_key = "pk-..." secret_key = "sk-..." base_url = "https://agent.goldebridge.com" tracing_enabled = true flush_at = 512 flush_interval = 5.0 sample_rate = 0.3 environment = "production" release = "1.0.0" [taotoken] api_key = "你的 TaoToken Key" base_url = "https://taotoken.net/api" default_model = "gpt-4"

读取时用tomllib(Python 3.11+)或toml库加载,这样不同环境切换只改配置文件,代码不动。

4. 验证请求:从 Trace 到指标,确认链路数据真的通了

配置完成后,第一步是验证数据能不能到平台。写一个最小可运行脚本:

import os from agentinsight import observe, get_client from openai import OpenAI llm_client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) @observe(as_type="generation", name="smoke-test-generation") def smoke_test(): response = llm_client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": "用一句话解释什么是可观测性。"}], ) return response.choices[0].message.content if __name__ == "__main__": result = smoke_test() print("模型返回:", result) client = get_client() client.flush() print("Trace 已发送,请到 AgentInsight 平台查看。")

运行后,如果平台能看到一条名为smoke-test-generation的 Trace,说明链路通了。如果看不到,先检查flush()有没有调用,再检查网络出口能不能访问agent.goldebridge.com。

验证通过后,把上下文传播加上,这样 Trace 能关联到具体用户和会话:

from agentinsight import AgentInsight, propagate_attributes client = AgentInsight() with client.start_as_current_observation(name="user-request", as_type="span") as root: with propagate_attributes( user_id="user_12345", session_id="session_abc", metadata={ "environment": "production", "variant": "prompt-v2", "feature": "rag-agent", }, tags=["production", "v2", "rag"], ): answer = run_rag_agent("用户的问题") client.flush()

跨服务传播也支持,通过 HTTP 头部(基于 OpenTelemetry Baggage 协议)在微服务之间传递上下文:

from agentinsight import propagate_attributes import requests with propagate_attributes( user_id="user_12345", session_id="session_abc", as_baggage=True, ): result = requests.post( "http://downstream-service/api/process", json=data, )

有了这些数据,你可以在平台上构建监控指标体系。核心指标分四类:延迟类(Trace 数量、平均耗时、P95/P99、首 Token 时间)、成本类(总成本、按项目拆分、按会话拆分、趋势对比)、模型类(各模型调用量、延迟分布、Token 用量、成本/调用)、错误类(错误率趋势、异常类型分布、失败 Trace 详情、智能预警)。

验证成功的结果是:你在平台上能按用户、会话、标签、元数据过滤 Trace,能下钻到某一条失败 Trace 看到具体是哪个 Span 报错、输入输出是什么、耗时多少。这时候可观测才算真正落地。

5. 常见报错排查:401、local proxy failed、reading choices、OAuth

接入过程中最容易卡在几个报错上,这里逐个对照排查。

401 Unauthorized:最常见的是 Key 配错。检查三处:AgentInsight 的public_key/secret_key是否和控制台一致;TaoToken 的 API Key 是否复制完整(注意前后空格);环境变量有没有被其他 shell 覆盖。如果用了.env文件,确认加载顺序,别让旧变量覆盖新变量。

local proxy failed / connection refused:这类报错通常是网络出口问题。先确认运行环境能不能访问https://taotoken.net/api和https://agent.goldebridge.com。如果是容器环境,检查 DNS 配置和出站规则。注意不要配置任何非官方的网络转发工具,直接用官方通道即可。如果公司网络有出口限制,联系运维放行这两个域名。

reading 'choices' of undefined:这个报错说明模型返回结构不对,通常是 Base URL 配错导致请求打到了非预期端点。检查base_url是不是https://taotoken.net/api,不要多加/v1。另外确认model参数是 TaoToken 支持的模型 ID,写错模型名也可能返回非标准结构。还有一种情况是流式响应没处理完就取choices,加个判空:

response = llm_client.chat.completions.create(...) if not response or not response.choices: raise ValueError("模型返回为空,检查 Base URL 和 Model ID") content = response.choices[0].message.content

OAuth / authentication failed:如果你用的是 Claude Code 或 Codex 这类工具,认证方式可能不是简单 API Key。以 Claude Code 为例,需要配置ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY,Base URL 指向 TaoToken 通道,Key 用 TaoToken 创建的 Key。Codex 的auth.json里要写全三件套:Base URL、Key、Model ID。如果出现 OAuth 相关报错,先确认是不是混用了官方登录态和 API Key 两种认证方式,二选一即可。

Trace 不显示 / Span 丢失:检查flush()有没有在请求结束前调用;检查sample_rate是不是设太低;检查@observe装饰的函数是不是真的被执行了(有些异步框架里装饰器位置不对会失效)。如果是异步场景,确认用的是支持 async 的装饰方式,或者在事件循环结束后手动 flush。

Token 统计为 0:AgentInsight 依赖模型返回的 usage 字段。如果 TaoToken 通道返回的响应里没有 usage,需要手动在gen_span.update()里补usage_details。另外确认模型 ID 在平台的定价表里能匹配到,否则成本计算会是 0。

排查顺序建议:先看 SDK 日志有没有发送成功,再看平台有没有收到,最后看指标计算对不对。大部分问题出在第一步和第二步之间,也就是网络和 Key 配置。

6. 从可观测到可优化:评分系统、实验框架与持续调优闭环

可观测和可监控告诉你“发生了什么”,可优化解决的是“怎么做得更好”。AgentInsight 支持三种评分类型,可以对每次 Agent 执行进行质量标记:

from agentinsight import observe, get_client from agentinsight.api.commons.types.score_data_type import ScoreDataType @observe(name="customer-service-agent") def handle_customer_query(query: str) -> str: answer = run_rag_agent(query) return answer result = handle_customer_query("如何退换货?") client = get_client() with client.start_as_current_observation( name="quality-evaluation", as_type="evaluator", ) as eval_span: eval_span.score( name="relevance", value=0.92, data_type=ScoreDataType.NUMERIC, comment="回答与问题高度相关", ) eval_span.score( name="hallucination_free", value=1.0, data_type=ScoreDataType.BOOLEAN, ) eval_span.score( name="sentiment", value="positive", data_type=ScoreDataType.CATEGORICAL, ) client.flush()

评分来源可以多样:人工评分适合运营/审核人员标注客服抽检打分;自动评分适合代码逻辑自动判断是否包含关键词、格式是否正确;LLM-as-Judge 适合用另一个 LLM 评估回答质量;用户反馈适合用户点赞/点踩按钮反馈。

有了评分,就可以跑实验框架做 A/B 测试:

from agentinsight import AgentInsight, Evaluation client = AgentInsight() def rag_task(*, item, **kwargs): input_data = item["input"] if isinstance(item, dict) else item.input return run_rag_agent(input_data) def relevance_evaluator(*, input, output, expected_output=None, **kwargs): if not expected_output: return Evaluation(name="relevance", value=0, comment="缺少期望输出") overlap = set(expected_output.split()) & set(output.split()) score = len(overlap) / max(len(expected_output.split()), 1) return Evaluation( name="relevance", value=round(score, 2), comment=f"关键词重合率: {score:.0%}", ) def cost_evaluator(*, input, output, expected_output=None, **kwargs): cost = len(output) * 0.00001 return Evaluation( name="cost_score", value=max(0, 1.0 - cost), comment=f"估算成本: ${cost:.4f}", ) result = client.run_experiment( name="rag-prompt-v2-vs-v1", data=[ {"input": "如何退换货?", "expected_output": "退换货流程 7天无理由 快递上门"}, {"input": "会员有什么权益?", "expected_output": "会员权益 折扣 积分 专属客服"}, {"input": "配送范围是什么?", "expected_output": "配送范围 全国 主要城市 次日达"}, ], task=rag_task, evaluators=[relevance_evaluator, cost_evaluator], ) for item_result in result.item_results: print(f"输入: {item_result.item}") print(f"输出: {item_result.output}") for evaluation in item_result.evaluations: print(f" {evaluation.name}: {evaluation.value} ({evaluation.comment})")

这个实验框架的价值在于:把“感觉 Prompt 效果变好了”变成“数据证明 Prompt v2 的相关性评分比 v1 高 15%”。持续优化闭环就是:采集 Trace + 评分 → 识别低分场景 → 设计改进方案(优化 Prompt / 更换模型 / 调整检索策略)→ 运行 A/B 实验 → 发布最优版本 → 回到第一步。

TypeScript 开发者也有完整 SDK。初始化用 OpenTelemetry 的 NodeSDK:

import { AgentInsightSpanProcessor } from "@agentinsight-sdk/otel"; import { NodeSDK } from "@opentelemetry/sdk-node"; const sdk = new NodeSDK({ spanProcessors: [ new AgentInsightSpanProcessor({ publicKey: process.env.AGENTINSIGHT_PUBLIC_KEY!, secretKey: process.env.AGENTINSIGHT_SECRET_KEY!, baseUrl: "https://agent.goldebridge.com", }), ], }); sdk.start();

OpenAI 集成用observeOpenAI包装:

import { observeOpenAI } from "@agentinsight-sdk/openai"; import OpenAI from "openai"; const client = observeOpenAI( new OpenAI({ apiKey: process.env.TAOTOKEN_API_KEY, baseURL: process.env.TAOTOKEN_BASE_URL, }), ); const result = await client.chat.completions.create({ model: "gpt-4", messages: [{ role: "user", content: "Hello!" }], });

生产环境还有几个最佳实践。采样策略上,开发/测试环境sample_rate=1.0全量采集,生产环境 0.1~0.3 采样,关键场景(如付费用户)通过propagate_attributes的 tags 标记确保被采集。数据脱敏方面,TypeScript 版 SDK 内置了脱敏功能,可以自动屏蔽 API Key、手机号、身份证号等敏感信息。多项目隔离上,每个应用用独立 API Key,数据完全隔离。批量发送默认flush_at=512、flush_interval=5s,高并发场景可以调到flush_at=1024、flush_interval=10.0,避免对业务性能产生影响。

调优验证动作建议固定成流程:每周跑一次实验对比,每次 Prompt 变更都带评分,每次模型切换都看延迟和成本曲线。这样 Agent 的优化就不是凭感觉,而是数据驱动。需要长期做编码和 Agent 调优的团队,可以用 Coding Plan 把通道和额度统一管理;只是验证模型效果的,直接用模型对话快速试;接入和排障阶段,API Keys 和接入文档是最常翻的两页。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询