Trace 体系完整设计与接入指南

from by_framework.trace import SpanRecord, TraceWriteClient

writer = TraceWriteClient()

await writer.record_span(
    SpanRecord(
        trace_id=header.trace_id,
        span_id="custom-step-1",
        parent_span_id=header.trace_parent_span_id,
        name="Knowledge_Retrieval",
        operation="knowledge.retrieve",
        component="external-worker",
        kind="retriever",
        start_ts=start_ms,
        end_ts=end_ms,
        input={"query": query},
        output={"documents": len(docs)},
        source="external",
        session_id=header.session_id,
        message_id=header.message_id,
    )
)

from by_framework.trace import EventRecord, TraceWriteClient

writer = TraceWriteClient()

await writer.record_event(
    EventRecord(
        event_id="evt-001",
        trace_id=header.trace_id,
        session_id=header.session_id,
        message_id=header.message_id,
        event_type="chunk",
        content_type="text",
        timestamp=now_ms,
        payload={"content": "partial answer"},
    )
)

from by_framework_trace_query import TraceReadClient

client = TraceReadClient()

by_session = await client.list_traces(session_id="sess-123", limit=20)
by_worker = await client.list_traces(worker_id="worker-1", limit=20)
by_agent = await client.list_traces(agent_type="weather-agent", limit=20)

from by_framework_trace_query import TraceReadClient

client = TraceReadClient(max_spans=1000)
explain = await client.explain_trace(trace_id, session_id=session_id)

# diagnostics codes:
# missing_client_dispatch, missing_worker_execute, missing_parent,
# parent_cycle, trace_output_missing, span_count_exceeded, source_timeout
print(explain)

class PlannerWorker(LangGraphWorker):
    def get_thread_id(self, context):
        # 推荐使用 message_id，避免同 session 下旧挂起状态串扰。
        return context.message_id or context.session_id

    def build_graph(self):
        # make_remote_agent_tool 会通过 context.call_agent 派发子 agent。
        # trace parent 会自动使用 agent.workflow observation。
        ...

# 时间语义：
# - A initial worker.execute: 执行到 interrupt / QUEUED
# - B worker.execute: 子 agent 实际耗时
# - A resume worker.execute: 回调后继续执行
# - agent.workflow:A: 覆盖完整逻辑流程

from langfuse import Langfuse
from by_framework.trace import (
    extract_external_trace_context,
    start_langfuse_observation,
    build_otel_parent_context,
)
from opentelemetry import trace

ctx = extract_external_trace_context(command_dict)
langfuse = Langfuse()

obs = start_langfuse_observation(
    langfuse,
    command_dict,
    name="external_plain_app",
    as_type="span",
    input_data=command_dict.get("content"),
)

tracer = trace.get_tracer("external-app")
with tracer.start_as_current_span(
    "external_plain_app",
    context=build_otel_parent_context(command_dict),
):
    result = run_external_pipeline()

obs.update(output=result)
obs.end()

# orchestrator_worker.py
import os
from typing import Any, List
from by_framework.worker import GatewayWorker, run_worker
from by_framework.core.protocol.commands import AskAgentCommand, ResumeCommand
from by_framework.worker.context import AgentContext

class OrchestratorWorker(GatewayWorker):

    def get_agent_types(self) -> List[str]:
        return ["orchestrator"]

    async def process_command(
        self, command: AskAgentCommand | ResumeCommand, context: AgentContext
    ) -> Any:

        if isinstance(command, ResumeCommand):
            # ── 恢复阶段：sub-agent 已返回 ──
            # command.reply_data 是子 agent 的结果
            weather_result = command.reply_data
            await context.emit_chunk(
                f"天气信息已获取：{weather_result}\n根据天气，建议携带雨具出行。"
            )
            return {"status": "done", "weather": weather_result}

        # ── 初始阶段：收到用户请求 ──
        # context.call_agent 自动完成：
        #   1. 在 Langfuse 创建 agent.call_agent:weather-agent span
        #   2. 将 langfuse_parent_observation_id 注入转发命令的 header
        #   3. 挂起本 worker，等待 ResumeCommand 返回
        await context.call_agent(
            target_agent_type="weather-agent",
            content={"city": "北京", "query": command.content},
            wait_for_reply=True,
        )

if __name__ == "__main__":
    run_worker(
        OrchestratorWorker,
        worker_id="orchestrator-worker-1",
        redis_host=os.getenv("BYAI_REDIS_HOST", "127.0.0.1"),
        redis_port=int(os.getenv("BYAI_REDIS_PORT", 6379)),
        plugins=[_make_langfuse_plugin()],
    )

# weather_agent_worker.py
import os
from typing import List
from by_framework.worker import run_worker
from by_framework_langgraph import LangGraphWorker
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

class WeatherAgentWorker(LangGraphWorker):

    def get_agent_types(self) -> List[str]:
        return ["weather-agent"]

    def build_graph(self, context, command):
        llm = ChatOpenAI(
            model=os.getenv("LLM_MODEL", "gpt-4o"),
            base_url=os.getenv("OPENAI_BASE_URL"),
            api_key=os.getenv("OPENAI_API_KEY"),
            streaming=True,
            # ↓ 必须开启，否则 streaming 模式下拿不到 token 用量
            stream_options={"include_usage": True},
        )
        return create_react_agent(
            llm,
            tools=[],
            checkpointer=self.get_checkpointer(),
            prompt="你是专业的天气助手，根据城市给出天气状况和出行建议，不超过100字。",
        )

if __name__ == "__main__":
    run_worker(
        WeatherAgentWorker,
        worker_id="weather-agent-worker-1",
        redis_host=os.getenv("BYAI_REDIS_HOST", "127.0.0.1"),
        redis_port=int(os.getenv("BYAI_REDIS_PORT", 6379)),
        plugins=[_make_langfuse_plugin()],
    )

# 两个 worker 都需要注入同一套 LangfusePlugin 配置
import os
from by_framework_trace_langfuse import LangfusePlugin, LangfuseConfig

def _make_langfuse_plugin() -> LangfusePlugin:
    return LangfusePlugin(
        config=LangfuseConfig(
            public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
            secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
            base_url=os.getenv("LANGFUSE_BASE_URL"),
        )
    )

# 在 run_worker 调用时传入：
# run_worker(MyWorker, ..., plugins=[_make_langfuse_plugin()])