Metadata-Version: 2.4
Name: ylg-open-trace-agent-sdk
Version: 0.2.1
Summary: OpenTelemetry-based trace SDK for Dockerized Python Agents.
Author: wangxiao
Keywords: agent,opentelemetry,sdk,trace,tracing
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.11
Requires-Dist: opentelemetry-api<2.0.0,>=1.28.0
Requires-Dist: opentelemetry-sdk<2.0.0,>=1.28.0
Description-Content-Type: text/markdown

# 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 安装：

```bash
uv add ylg-open-trace-agent-sdk
```

也可以使用 pip 安装：

```bash
pip install ylg-open-trace-agent-sdk
```

代码中导入的 Python 包名是：

```python
from ylg_open_trace_agent_sdk import TraceClient, TraceContext
```

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

```python
from ylg_open_trace_agent_sdk import agent_progress_event, agent_result_event, answer_token_event
```

## 基础配置

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

```text
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：

```python
trace = TraceClient.from_env(use_global_provider=True)
```

## 最小接入

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

```python
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)
```

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

```python
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 流程按下面方式分层：

```text
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。

```python
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 阶段

```python
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 阶段

```python
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

```python
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，并继续抛出原始异常。

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

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

### 记录事件和异常

```python
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、命中来源，不记录完整原文。
- 对外部调用记录结果数量、状态码、业务错误码，避免记录完整响应体。

## 测试

普通单测：

```bash
uv run pytest
```

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

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