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