learnNote

主题

7. 自定义 Trace 扩展 ​

1. 适用场景 ​

标准 LlmAgent 的埋点由 _llm_processor.py / _tools_processor.py 自动完成。以下场景需额外或替代埋点：

场景方案
RemoteA2AAgent、ClaudeAgent 等CustomTraceReporter
LangGraph Agent_langgraph.py 内置 trace
用户自定义 Tool / Agent直接使用 tracer + Langfuse attributes
业务自定义步骤tracer.start_as_current_span() + langfuse.observation.*

2. CustomTraceReporter ​

位于 trpc_agent_sdk/telemetry/_custom_trace.py，封装 Event 流 → Span 的转换逻辑。

基本用法 ​

python
from trpc_agent_sdk.telemetry import CustomTraceReporter

class MyCustomAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext):
        reporter = CustomTraceReporter(
            agent_name=self.name,
            model_prefix="my_custom",           # trace 中 model = "my_custom:{agent_name}"
            tool_description_prefix="My tool",
        )

        async for event in self._process_events():
            reporter.trace_event(ctx, event)
            yield event

trace_event() 逻辑 ​

Event 类型行为
partial=True跳过
含 function_call记入 pending_function_calls，等待 response
含 function_response创建 execute_tool Span + trace_tool_call()
含完整 text创建 call_llm Span + trace_call_llm()

Instruction 关联 ​

python
instruction = getattr(ctx.agent, 'instruction', None)
instruction_metadata = getattr(instruction, 'metadata', None)
trace_call_llm(..., instruction_metadata=instruction_metadata)

与标准 LlmAgent 行为一致。

3. LangGraph Agent ​

agents/utils/_langgraph.py 在 LangGraph 节点执行时：

python
with tracer.start_as_current_span("call_llm"):
    trace_call_llm(ctx, Event.new_id(), llm_request, llm_response)

with tracer.start_as_current_span(f"execute_tool {actual_func.__name__}"):
    trace_tool_call(...)

不经过 _llm_processor，但产出相同的 gen_ai.operation.name，映射层行为一致。

4. 用户自定义 Span（官方推荐写法） ​

python
from trpc_agent.telemetry import tracer
from opentelemetry import trace

def my_tool_logic() -> str:
    with tracer.start_as_current_span("my_custom_step"):
        span = trace.get_current_span()
        span.set_attribute("gen_ai.operation.name", "custom_step")  # 可选
        span.set_attribute("langfuse.observation.input", "input data")
        span.set_attribute("langfuse.observation.output", "output data")
    return "done"

未识别的 gen_ai.operation.name 会走默认映射：保留 attributes + langfuse.observation.type=span。

也可直接写 Langfuse 语义属性，映射层在 default 分支会 span_attrs.update(attributes) 透传。

5. A2A 场景的 Trace 上下文 ​

server/a2a/executor/_a2a_agent_executor.py 在 A2A 请求处理时：

python
from opentelemetry.context import attach
from opentelemetry.propagate import extract

token = attach(extract(headers))  # 从 HTTP headers 恢复 OTel context

使 Agent 内部 Span 可关联到上游传入的 trace context。配合 enable_a2a_trace=True 可保留 a2a-sdk Span。

6. Debug Server 的 Span 名称 ​

Debug Web UI 会调用 set_trpc_agent_span_name() 修改属性前缀（默认 trpc.python.agent），以便 UI 展示。Langfuse 映射使用 get_trpc_agent_span_name() 读取当前前缀，与 Debug 模式兼容。

7. 扩展检查清单 ​

新增自定义 Agent 类型时，确认：

  • [ ] LLM 调用是否有 call_llm Span + trace_call_llm()
  • [ ] Tool 调用是否有 execute_tool Span + trace_tool_call()
  • [ ] 是否设置 gen_ai.operation.name
  • [ ] 是否写入 token usage（gen_ai.usage.*）
  • [ ] Instruction metadata 是否传递（若使用 RemoteInstructionManager）
  • [ ] 是否在 langfuse_setup() 之后执行

8. 与 cc_to_langfuse 扩展性对比 ​

cc_to_langfusetrpc-agent
扩展方式修改 parser / emit 逻辑添加 OTel Span + attributes
新 Observation 类型调用 SDK span() / generation()设 langfuse.observation.type
子代理读 agent-*.jsonl 嵌套 emitAgentTool / transfer_to_agent 走框架 Span
侵入性零侵入 CC需在 Agent 代码或框架层埋点

tRPC-Agent 的扩展点是 OpenTelemetry 标准，任何 OTel 生态工具都能消费同一套 Span。