ylg-open-trace-agent-sdk 0.2.0

OpenTelemetry-based trace SDK for Dockerized Python Agents.

YLG Open Trace Agent SDK

YLG Open Trace Agent SDK 是面向独立 Docker Python Agent 的运行链路 SDK。它基于 OpenTelemetry 管理 span 生命周期，并把 Agent 运行过程中的 workflow、tool、RAG、LLM、MCP 等节点转换为平台 /api/v2/traces/ingest 可接收的 trace events。

安装

使用 uv 安装：

uv add ylg-open-trace-agent-sdk

也可以使用 pip 安装：

pip install ylg-open-trace-agent-sdk

代码中导入的 Python 包名是：

from ylg_open_trace_agent_sdk import TraceClient, TraceContext

Streaming HTTP Agent 需要输出平台 SSE 时，可单独使用 streaming helper：

from ylg_open_trace_agent_sdk import agent_progress_event, agent_result_event, answer_token_event

基础配置

生产环境建议通过环境变量配置平台 trace ingest 地址：

OPEN_TRACE_AGENT_ENDPOINT=http://127.0.0.1:8000/api/v2/traces/ingest
OPEN_TRACE_AGENT_SERVICE_NAME=road_advice_agent
OPEN_TRACE_AGENT_FLUSH_INTERVAL_MS=1000
OPEN_TRACE_AGENT_MAX_QUEUE_SIZE=1000

字段说明：

• OPEN_TRACE_AGENT_ENDPOINT：必填，平台 trace ingest 完整 URL。
• OPEN_TRACE_AGENT_SERVICE_NAME：可选，写入 OpenTelemetry service.name。
• OPEN_TRACE_AGENT_FLUSH_INTERVAL_MS：可选，后台批量上报间隔，默认 1000。
• OPEN_TRACE_AGENT_MAX_QUEUE_SIZE：可选，内存队列上限，默认 1000。

如果 Agent 框架已经初始化了全局 OpenTelemetry provider，例如 Agno/OpenInference，可复用全局 provider：

trace = TraceClient.from_env(use_global_provider=True)

最小接入

每次 Agent 执行都应该先创建一个 TraceContext，然后用 agent_run 包住完整运行过程。子节点必须在 agent_run 内创建，否则 SDK 无法知道当前 run 的平台上下文。

from ylg_open_trace_agent_sdk import TraceClient, TraceContext

trace = TraceClient.from_env()

context = TraceContext(
    trace_id="trace-001",
    request_id="request-001",
    run_id="run-001",
    agent_id="road_advice_agent",
    agent_version="v1",
    tenant_id="tenant-a",
    product_id="road-decision",
    scene_code="maintenance-advice",
)

with trace.agent_run(context, name="road_advice_agent", input={"question": "这段路怎么养护？"}):
    with trace.tool("condition_api.query", input={"section_id": "G42-K121"}, attributes={"tool_type": "datastore"}) as span:
        result = {"row_count": 3}
        span.set_output(result)

    with trace.llm(
        "final_answer",
        input={"prompt": "生成养护建议"},
        attributes={"provider_name": "openai", "request_model": "gpt-5.2"},
    ) as span:
        span.set_token_usage(input_tokens=120, output_tokens=48)
        span.set_output({"answer": "建议进行预防性养护。"})

trace.flush(timeout_ms=3000)

服务进程退出前建议调用：

trace.close()

建议埋点位置

建议把 trace 加在“能解释一次 Agent 决策为什么这样产生”的关键路径上，而不是给每一行代码都加 span。

流程位置	推荐 API	建议记录内容
Agent 入口	trace.agent_run(...)	平台上下文、用户问题、run 输入摘要
任务规划、状态机阶段、LangGraph 节点	trace.workflow(...) 或 trace.custom(..., span_kind="langgraph.node")	阶段名称、输入摘要、阶段输出
外部 API、数据库、文件、规则引擎调用	trace.tool(...)	tool 名称、查询条件摘要、返回数量、耗时相关指标
RAG 查询改写	trace.rag(..., stage="retrieval.query_rewrite")	原始 query、改写 query、改写策略
RAG 召回	trace.rag(..., stage="retrieval.recall")	数据源 ID、召回数量、top-k、过滤条件
RAG 重排	trace.rag(..., stage="retrieval.rerank")	reranker 名称、候选数量、保留数量
RAG grounding/citation	trace.rag(..., stage="retrieval.grounding")	引用文档 ID、证据片段摘要
LLM 调用	trace.llm(...)	provider、request model、response model、token 用量、输出摘要
MCP 调用	trace.mcp(...)	server/tool 名称、参数摘要、返回状态
产物生成、数据契约校验、业务分支判断	trace.custom(...)	产物 ID、校验结果、分支原因

推荐一条完整 Agent 流程按下面方式分层：

agent_run
  workflow: plan
  workflow: retrieve_context
    rag: retrieval.query_rewrite
    rag: retrieval.recall
    rag: retrieval.rerank
    rag: retrieval.grounding
  workflow: execute_tools
    tool: condition_api.query
    tool: policy_api.query
  workflow: generate_answer
    llm: final_answer
  custom: artifact

常用写法

输出平台 streaming event

TraceClient 只负责 /api/v2/traces/ingest 链路观测；answer_token、agent_progress、agent_result 和 agent_error 应通过远端 Agent 的 /invoke/stream SSE response 输出，不能混进 trace ingest。

from ylg_open_trace_agent_sdk import agent_progress_event, agent_result_event, answer_token_event


async def stream_answer(deltas):
    yield agent_progress_event(node="generation", phase="start", message="开始生成回答").to_sse()
    chunks = []
    async for index, delta in deltas:
        chunks.append(delta)
        yield answer_token_event(content=delta, index=index).to_sse()
    answer = "".join(chunks)
    yield agent_progress_event(node="generation", phase="end", message="回答生成完成").to_sse()
    yield agent_result_event(assistant_text=answer).to_sse()

平台只透传远端真实 answer_token，不会默认把最终 assistant_text 切成伪 token；最终消息应以 run_done.output.assistant_text / agent_result.assistant_text 为准。

记录 workflow 阶段

with trace.workflow("retrieve_context", input={"query": question}) as span:
    docs = retrieve_docs(question)
    span.set_metrics({"documents": len(docs)})
    span.set_output({"document_ids": [doc["id"] for doc in docs]})

记录 RAG 阶段

with trace.rag(
    "policy_index",
    input={"query": question, "top_k": 8},
    stage="retrieval.recall",
    attributes={"data_source_id": "policy-vdb"},
) as span:
    docs = vector_store.search(question, top_k=8)
    span.set_metrics({"documents": len(docs)})
    span.set_output({"document_ids": [doc["id"] for doc in docs]})

记录 LLM 调用和 token

with trace.llm(
    "generate_answer",
    input={"messages": messages},
    attributes={"provider_name": "openai", "request_model": "gpt-5.2"},
) as span:
    response = llm.chat(messages)
    span.set_token_usage(
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        reasoning_output_tokens=getattr(response.usage, "reasoning_output_tokens", None),
    )
    span.set_output({"answer": response.text})

用装饰器包裹 Tool

同步和 async Tool 都可以用 instrument_tool 快速包裹。被包裹函数抛错时，SDK 会把对应 span 标记为 failed，并继续抛出原始异常。

@trace.instrument_tool("condition_api.query")
def query_condition(section_id: str) -> dict:
    return condition_api.query(section_id)

@trace.instrument_tool("policy_api.search")
async def search_policy(query: str) -> list[dict]:
    return await policy_api.search(query)

记录事件和异常

with trace.tool("condition_api.query", input={"section_id": section_id}) as span:
    span.event("request_started", payload={"section_id": section_id})
    try:
        result = condition_api.query(section_id)
    except TimeoutError as exc:
        span.record_exception(exc, error_code="CONDITION_API_TIMEOUT")
        raise
    span.set_output({"row_count": len(result)})

如果异常发生在 with 内且没有被业务代码捕获，SDK 会自动记录异常并把 span 标记为 failed。

数据与安全建议

input、set_input、set_output 默认会生成摘要，不会直接上传完整对象。摘要会截断长字符串，并对 password、token、secret、authorization、api_key、apikey 等字段脱敏。

接入时仍建议遵守以下规则：

• 不要把原始密钥、完整用户隐私、完整大文档直接放进 attributes。
• attributes 适合放可过滤、可聚合的短字段，例如 tool_type、data_source_id、provider_name、request_model。
• 大模型 prompt 和检索文档建议只记录摘要、ID、数量、top-k、命中来源，不记录完整原文。
• 对外部调用记录结果数量、状态码、业务错误码，避免记录完整响应体。

测试

普通单测：

uv run pytest

真实平台 smoke 需要先启动 Agent 平台 API，并让它连接 PostgreSQL 测试库：

OPEN_TRACE_AGENT_LIVE_BASE_URL=http://127.0.0.1:8000/api/v2 uv run pytest tests/test_live_platform_smoke.py -q