从可观测到可优化：用 AgentInsight SDK 打造 AI Agent 全链路监控与持续调优体系

导读：AI Agent 应用从 Demo 到生产，最大的鸿沟不是模型能力，而是"看不见、管不住、改不了"。本文以 AgentInsight 开源 SDK 为工具，从**可观测（Observability）→ 可监控（Monitoring）→ 可优化（Optimization）**三个层次，手把手演示如何为 AI Agent 构建一套完整的工程化治理体系。所有代码均基于 AgentInsight Python/TypeScript SDK v0.1.x，可直接运行。

一、为什么 Agent 需要"全链路工程化"

做过 Agent 应用的人大概都经历过这样的场景：

用户反馈"AI 回答不对"，你打开日志系统，看到的只有一行 POST /api/chat 200 OK。至于这中间模型调了几次、用了什么 Prompt、检索了哪些文档、工具返回了什么、Token 花了多少——全靠猜。

这不是个别现象，而是 Agent 应用在生产环境中的系统性工程缺失。

传统 Web 应用有 APM（Application Performance Monitoring）来覆盖接口耗时、错误率、资源使用等指标。但 Agent 应用的核心问题完全不同：

维度	传统 Web 应用	AI Agent 应用
执行链路	请求→处理→响应（固定）	规划→检索→推理→工具→校验→生成（动态）
失败模式	异常栈、HTTP 状态码	幻觉、格式错误、工具选错、上下文漂移
成本结构	服务器固定成本	Token 动态成本，随链路复杂度非线性增长
质量评估	单元测试、集成测试	需要持续评估输出质量、工具准确率、检索相关性

AgentInsight 正是为解决这些问题而生的开源 SDK + 企业级平台。它基于 OpenTelemetry 协议设计，支持 Python 和 TypeScript 两种语言，可以零侵入地接入你的 Agent 应用，采集 Trace 链路、模型调用、Token 消耗、响应耗时、异常错误等关键数据。

本文不重复讨论"为什么 Agent 需要可观测"（如果你感兴趣，可以参考之前的文章），而是直接进入工程实践：如何用 AgentInsight SDK 一步步搭建可观测、可监控、可优化的 Agent 工程体系。

---

二、可观测：让 Agent 执行链路透明化

可观测是一切的基础。只有先"看见"，才能谈"监控"和"优化"。

2.1 安装与初始化

# Python SDK 安装（Python >= 3.10）
pip install agentinsight-sdk

# 如果需要 OpenAI 集成
pip install agentinsight-sdk openai

# 如果需要 LangChain 集成
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"

from agentinsight import AgentInsight

client = AgentInsight()

方式二：代码直接初始化（适合快速验证）

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,           # 采样率 1.0 = 全量采集
    environment="production",  # 环境标识
    release="1.0.0",           # 版本标识
)

2.2 用 @observe 装饰器构建 Trace 链路

AgentInsight 最核心的能力是 @observe 装饰器——它可以自动追踪函数调用，并在嵌套调用时自动建立父子关系。

来看一个真实的 RAG Agent 场景：

from agentinsight import observe

@observe(as_type="agent", name="rag-agent")
def run_rag_agent(query: str) -> str:
    """RAG Agent 主入口"""
    # Step 1: 检索相关文档
    docs = retrieve_documents(query)

    # Step 2: 基于 Prompt 模板生成回答
    answer = generate_answer(query, docs)

    # Step 3: 安全校验
    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 都会自动记录：函数名、输入/输出、开始/结束时间、执行耗时、异常信息。

2.3 观察类型（Observation Types）

AgentInsight 定义了多种语义化的观察类型，对应 Agent 执行链路中的不同角色：

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

正确使用这些类型，可以让 Trace 在平台上呈现更清晰的语义结构，便于团队协作时快速理解链路含义。

2.4 手动管理 Span（高级场景）

对于需要更精细控制的场景（如异步调用、动态 Span 创建），可以使用底层 API：

from agentinsight import AgentInsight

client = AgentInsight()

with client.start_as_current_observation(
    name="process-user-query",
    as_type="span",
) as root_span:
    # 动态创建子 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("用户的具体问题")

        # 手动更新 Span 的输出和元数据
        gen_span.update(
            output=response,
            usage_details={
                "prompt_tokens": 150,
                "completion_tokens": 300,
            },
            cost_details={"total_cost": 0.0045},
        )

# 确保数据发送到服务端
client.flush()

这种方式适合需要精确控制 Span 生命周期或在运行时动态添加元数据的场景。

---

三、可监控：从 Trace 到指标体系

可观测解决的是"看见链路"，可监控解决的是"持续感知系统状态"。

3.1 OpenAI 一行接入——自动采集所有调用数据

如果你使用 OpenAI SDK，AgentInsight 提供了最简单的接入方式——只改一行 import：

# 之前
import openai

# 之后（仅修改 import 路径）
from agentinsight.openai import openai

# 后续代码完全不变
client = openai.OpenAI()

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "你是一个专业的技术助手。"},
        {"role": "user", "content": "解释一下什么是 RAG。"},
    ],
)

print(response.choices[0].message.content)

就这么简单。AgentInsight 会自动采集：

• 所有 Prompt 和 Completion 内容（包括 streaming、function calling）
• 请求延迟（首 Token 时间、总耗时）
• API 错误（超时、限流、模型不可用等）
• Token 用量（prompt_tokens、completion_tokens）
• 调用成本（基于模型定价自动计算）

这意味着你不需要修改任何业务逻辑，就能获得完整的模型调用监控。

3.2 LangChain 集成——覆盖复杂 Agent 链路

对于使用 LangChain 构建的 Agent 应用，AgentInsight 提供了 CallbackHandler：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from agentinsight.langchain import CallbackHandler

# 创建 AgentInsight 回调处理器
handler = CallbackHandler()

# 构建 LangChain 链
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的技术助手，请用中文回答。"),
    ("human", "{input}"),
])
chain = prompt | llm | StrOutputParser()

# 执行时传入 handler
result = chain.invoke(
    {"input": "什么是 Agent 可观测？"},
    config={"callbacks": [handler]},
)

CallbackHandler 会自动追踪 LangChain 链路中的每一个环节：Prompt 模板渲染、LLM 调用、Output Parser 处理等。

3.3 上下文传播——关联用户与会话

在生产环境中，Trace 不能是孤立的。你需要知道：这条 Trace 属于哪个用户、哪个会话、哪个业务场景。

AgentInsight 提供了 propagate_attributes 来实现上下文传播：

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",  # A/B 测试变体
            "feature": "rag-agent",
        },
        tags=["production", "v2", "rag"],
    ):
        # 所有嵌套调用都会自动携带这些上下文
        answer = run_rag_agent("用户的问题")

client.flush()

这样在 AgentInsight 平台上，你可以按用户、会话、标签、元数据等维度过滤和分析 Trace。

跨服务传播也支持——通过 HTTP 头部（基于 OpenTelemetry Baggage 协议）在微服务之间传递上下文：

from agentinsight import propagate_attributes

# 设置 baggage，自动通过 HTTP 头传播到下游服务
with propagate_attributes(
    user_id="user_12345",
    session_id="session_abc",
    as_baggage=True,
):
    # 这里的 HTTP 请求会自动携带上下文头
    result = requests.post("http://downstream-service/api/process", json=data)

3.4 构建监控指标体系

有了上述数据采集能力，你可以在 AgentInsight 平台上构建完整的监控指标体系：

┌─────────────────────────────────────────────────────┐
│                  AgentInsight 平台                   │
├──────────┬──────────┬──────────┬───────────────────┤
│ 研发仪表盘 │ 成本分析   │ 模型使用   │ 错误与异常         │
├──────────┼──────────┼──────────┼───────────────────┤
│ Trace 数量 │ 总成本     │ 各模型调用  │ 错误率趋势         │
│ 平均耗时   │ 按项目拆分  │ 延迟分布   │ 异常类型分布       │
│ P95/P99  │ 按会话拆分  │ Token 用量 │ 失败 Trace 详情    │
│ 吞吐量    │ 趋势对比   │ 成本/调用  │ 智能预警           │
└──────────┴──────────┴──────────┴───────────────────┘

---

四、可优化：基于数据的持续改进

可观测和可监控告诉你"发生了什么"，可优化解决的是"怎么做得更好"。

4.1 评分系统——量化 Agent 质量

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:
    """客服 Agent"""
    answer = run_rag_agent(query)
    return answer

# 执行 Agent
result = handle_customer_query("如何退换货？")

# 获取客户端并添加评分
client = get_client()

with client.start_as_current_observation(name="quality-evaluation", as_type="evaluator") as eval_span:
    # 数值评分（0-1）：回答相关性
    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 评估	GPT-4 评估回答质量
用户反馈	用户点赞/点踩	👍/👎 按钮反馈

4.2 实验框架——A/B 测试 Prompt 和模型

AgentInsight 内置了实验框架，可以系统性地对比不同 Prompt 版本、不同模型、不同参数配置的效果：

from agentinsight import AgentInsight, Evaluation

client = AgentInsight()

# 定义 Agent 任务
def rag_task(*, item, **kwargs):
    """被测试的 Agent 任务"""
    input_data = item["input"] if isinstance(item, dict) else item.input
    # 这里是你的 Agent 逻辑
    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="缺少期望输出")
    # 简单的关键词匹配（实际中可用 LLM-as-Judge）
    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%"。

4.3 基于评分的持续优化闭环

结合评分系统和实验框架，可以建立一个完整的持续优化闭环：

┌──────────────────────────────────────────────────────┐
│                                                      │
│   ① 采集 Trace + 评分                                 │
│        │                                              │
│        ▼                                              │
│   ② 识别低分场景                                      │
│      （哪些问题类型评分低？哪些工具调用失败率高？）        │
│        │                                              │
│        ▼                                              │
│   ③ 设计改进方案                                      │
│      （优化 Prompt / 更换模型 / 调整检索策略）          │
│        │                                              │
│        ▼                                              │
│   ④ 运行 A/B 实验                                     │
│      （对比改进前后的效果）                             │
│        │                                              │
│        ▼                                              │
│   ⑤ 发布最优版本 → 回到 ①                              │
│                                                      │
└──────────────────────────────────────────────────────┘

这个闭环的核心思想是：Agent 的优化不应该是凭感觉的，而应该是数据驱动的。

---

五、TypeScript/Node.js 开发者的选择

AgentInsight 同样提供了完整的 TypeScript SDK，API 设计与 Python 版保持一致。

5.1 初始化

import { AgentInsightSpanProcessor } from "@agentinsight-sdk/otel";
import { NodeSDK } from "@opentelemetry/sdk-node";

// 初始化 OpenTelemetry + AgentInsight
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();

5.2 OpenAI 集成

import { observeOpenAI } from "@agentinsight-sdk/openai";
import OpenAI from "openai";

// 用 observeOpenAI 包装 OpenAI 客户端
const client = observeOpenAI(
  new OpenAI({ apiKey: process.env.OPENAI_API_KEY }),
);

// 后续调用自动追踪
const result = await client.chat.completions.create({
  model: "gpt-4",
  messages: [{ role: "user", content: "Hello!" }],
});

5.3 observe() 高阶函数

import { observe } from "@agentinsight-sdk/tracing";

// 用 observe 包装异步函数，自动追踪
const processOrder = observe(
  async (orderId: string, items: CartItem[]) => {
    const validation = await validateOrder(orderId, items);
    const payment = await processPayment(validation);
    return { orderId, status: "confirmed" };
  },
  {
    name: "process-order",
    asType: "span",
    captureInput: true,
    captureOutput: true,
  },
);

// 调用方式不变
const result = await processOrder("ord_123", cartItems);

5.4 LangChain 集成

import { CallbackHandler } from "@agentinsight-sdk/langchain";
import { ChatOpenAI } from "@langchain/openai";

const handler = new CallbackHandler({
  sessionId: "session-123",
  tags: ["production"],
});

const model = new ChatOpenAI({
  callbacks: [handler],
});

---

六、生产环境最佳实践

6.1 采样策略

不是所有 Trace 都需要全量采集。在流量较大的生产环境中，可以配置采样率：

# 环境变量方式
export AGENTINSIGHT_SAMPLE_RATE=0.3  # 采集 30% 的 Trace

# 或代码方式
client = AgentInsight(sample_rate=0.3)

建议策略：

• 开发/测试环境：sample_rate=1.0（全量采集）
• 生产环境：sample_rate=0.1~0.3（采样采集）
• 关键场景（如付费用户）：通过 propagate_attributes 的 tags 标记，确保关键请求被采集

6.2 数据脱敏

Agent 应用经常处理用户敏感数据。AgentInsight SDK（TypeScript 版）内置了数据脱敏功能，可以自动掩码 API Key、手机号、身份证号等敏感信息。

6.3 多项目隔离

如果你的团队管理多个 AI 应用，AgentInsight 支持多项目隔离——每个应用使用独立的 API Key，数据完全隔离，互不影响。

6.4 批量发送与性能

AgentInsight SDK 默认使用批量发送策略（flush_at=512，flush_interval=5s），避免对业务性能产生影响。在高并发场景下，可以调整参数：

client = AgentInsight(
    flush_at=1024,        # 更大的批量
    flush_interval=10.0, # 更长的间隔
)

---

七、总结

AI Agent 的工程化治理不是可选项，而是从 Demo 走向生产的必经之路。

本文展示的三个层次构成了一个完整的闭环：

层次	目标	AgentInsight 能力
可观测	看见链路	@observe 装饰器、手动 Span、多种观察类型
可监控	持续感知	OpenAI/LangChain 一行接入、上下文传播、平台仪表盘
可优化	持续改进	评分系统、实验框架、数据驱动优化闭环

核心观点：Agent 的优化不应该是凭感觉的，而应该是数据驱动的。只有建立了完整的可观测体系，才能从"猜测问题"走向"定位问题"，从"感觉变好了"走向"数据证明变好了"。

AgentInsight SDK 开源地址：

• Python SDK：github.com/AgentInsight/agentinsight-sdk-python
• TypeScript SDK：github.com/AgentInsight/agentinsight-sdk-ts
• 官网：agentinsight.goldebridge.com

---

相关阅读：

• Agent 开发的真正痛点：不是调用模型，而是把 Agent 稳定跑进生产
• 从 Demo 到生产：AI Agent 可观测到底要看什么？
• 我们开源了 AgentInsight：面向 LangChain / OpenAI 的 AI Agent 可观测 SDK