From 214a5982034ed23c73ca28510b03f326ede9fc90 Mon Sep 17 00:00:00 2001 From: "dajun.cui" Date: Wed, 24 Jun 2026 15:11:26 +0800 Subject: [PATCH] feat: add LangSmith OTel observability example for AgentKit APMPlus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add langchain_otel_agent.py: LangGraph multi-step agent with LangSmith OTel auto-tracing, exporting traces to AgentKit APMPlus via OTLP gRPC - Add langchain_otel_guide.md: integration guide for the observability approach - Update requirements.txt: add OTel and LangSmith dependencies Key features: - Zero-invasive tracing via OtelSpanProcessor on existing TracerProvider - APMPlus integration with gRPC header auth (x-byteapm-appkey) - Resource attribute: apmplus.business_carrier=agentkit_runtime - FastAPI auto-instrumentation for HTTP-level spans - Multi-path workflow demo (classify → simple_qa / research+tools / escalation) Co-Authored-By: Claude --- .../langchain_otel_agent.py | 335 ++++++++++++++++++ .../langchain_otel_guide.md | 131 +++++++ .../requirements.txt | 8 + 3 files changed, 474 insertions(+) create mode 100644 01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_agent.py create mode 100644 01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_guide.md diff --git a/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_agent.py b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_agent.py new file mode 100644 index 00000000..ef844000 --- /dev/null +++ b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_agent.py @@ -0,0 +1,335 @@ +import os +import json +import logging +from typing import TypedDict, Annotated, Literal +from langchain_openai import ChatOpenAI +from langchain_core.tools import tool +from langchain_core.messages import HumanMessage, SystemMessage +from langgraph.graph import StateGraph, START, END +from langgraph.graph.message import add_messages +from langgraph.prebuilt import ToolNode +from dotenv import load_dotenv +from fastapi import FastAPI, Request +from fastapi.responses import StreamingResponse +import uvicorn + +# OpenTelemetry — 最少侵入式接入 +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.resources import Resource +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor + +load_dotenv() +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + + +# ============ 0. OpenTelemetry 初始化（仅需这几行） ============ +OTEL_ENDPOINT = os.getenv("OBSERVABILITY_OPENTELEMETRY_APMPLUS_ENDPOINT", "http://localhost:4317") +OTEL_API_KEY = os.getenv("OBSERVABILITY_OPENTELEMETRY_APMPLUS_API_KEY", "") +SERVICE_NAME = os.getenv("OTEL_SERVICE_NAME", "langchain-agent") + +resource = Resource.create({ + "service.name": SERVICE_NAME, + "apmplus.business_carrier": "agentkit_runtime", +}) + +# 配置 OTLP exporter，header 中带上 ByteAPM AppKey +# 注意：gRPC metadata key 必须全小写 +exporter_headers = {} +if OTEL_API_KEY: + exporter_headers["x-byteapm-appkey"] = OTEL_API_KEY + +tracer_provider = TracerProvider(resource=resource) +tracer_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter( + endpoint=OTEL_ENDPOINT, + insecure=True, + headers=exporter_headers, + )) +) +trace.set_tracer_provider(tracer_provider) +tracer = trace.get_tracer(__name__) + +# LangSmith OTel 自动追踪（自动为所有 LangChain/LangGraph 调用生成 span） +# 无需手动在每个节点加 tracer，LangGraph 的每个节点、LLM 调用、工具调用都会自动生成子 span +try: + from langsmith.integrations.otel import configure as configure_langsmith_otel + configure_langsmith_otel( + project_name=os.getenv("LANGSMITH_PROJECT", SERVICE_NAME), + ) + logger.info("LangSmith OTel auto-tracing enabled") +except ImportError: + logger.warning("langsmith not installed, skipping auto-tracing. pip install langsmith") +except Exception as e: + logger.warning(f"LangSmith OTel configure failed: {e}, falling back to manual tracing only") + + +# ============ 1. 定义状态 ============ +class AgentState(TypedDict): + messages: Annotated[list, add_messages] + classification: str + final_answer: str + + +# ============ 2. 定义工具 ============ +@tool +def search_knowledge_base(query: str) -> str: + """搜索内部知识库获取相关信息""" + knowledge = { + "退款": "退款政策：7天无理由退款，需提供订单号。处理时间3-5个工作日。", + "配送": "标准配送3-5天，加急配送1-2天。满99元免运费。", + "账户": "账户问题请提供注册邮箱，支持修改密码、更换手机号、注销账户。", + } + for key, value in knowledge.items(): + if key in query: + return value + return "未找到相关信息，建议转人工客服。" + + +@tool +def check_order_status(order_id: str) -> str: + """查询订单状态""" + return f"订单 {order_id} 状态：已发货，预计明天送达。快递单号: SF1234567890" + + +@tool +def create_ticket(summary: str, priority: str) -> str: + """创建工单，用于需要人工跟进的问题""" + return f"工单已创建 - 摘要: {summary}, 优先级: {priority}, 工单号: TK-2024-001" + + +# ============ 3. LLM 和工具绑定 ============ +llm = ChatOpenAI( + model=os.getenv("MODEL_NAME", "gpt-4o-mini"), + api_key=os.getenv("OPENAI_API_KEY"), + base_url=os.getenv("OPENAI_API_BASE"), + temperature=0, +) + +tools = [search_knowledge_base, check_order_status, create_ticket] +llm_with_tools = llm.bind_tools(tools) + + +# ============ 4. 定义各节点（无需手动加 span，LangSmith OTel 自动追踪） ============ +def classify_intent(state: AgentState) -> AgentState: + """意图分类节点""" + messages = state["messages"] + response = llm.invoke([ + SystemMessage(content="""对用户问题进行分类，只返回以下类别之一： + - simple_qa: 简单问答，知识库可直接回答 + - order_query: 订单查询相关 + - complex_issue: 复杂问题，需要人工介入 + 回复格式：只返回类别名称"""), + messages[-1] + ]) + return {"classification": response.content.strip()} + + +def simple_qa_node(state: AgentState) -> AgentState: + """简单问答：查知识库直接回答""" + user_msg = state["messages"][-1].content + result = search_knowledge_base.invoke({"query": user_msg}) + response = llm.invoke([ + SystemMessage(content="基于以下知识库信息，友好地回答用户问题。"), + HumanMessage(content=f"知识库信息：{result}\n\n用户问题：{user_msg}") + ]) + return {"messages": [response], "final_answer": response.content} + + +def research_node(state: AgentState) -> AgentState: + """调研节点：调用工具收集信息""" + response = llm_with_tools.invoke([ + SystemMessage(content="你是客服助手，使用工具收集解决问题所需的信息。"), + *state["messages"] + ]) + return {"messages": [response]} + + +tool_node = ToolNode(tools) + + +def synthesize_node(state: AgentState) -> AgentState: + """综合节点：整合所有信息生成最终回答""" + response = llm.invoke([ + SystemMessage(content="""基于收集到的所有信息，生成最终回答。要求： + 1. 清晰总结问题处理结果 + 2. 如有后续步骤，明确告知用户 + 3. 语气友好专业"""), + *state["messages"] + ]) + return {"messages": [response], "final_answer": response.content} + + +def escalation_node(state: AgentState) -> AgentState: + """升级节点：创建工单转人工""" + user_msg = state["messages"][0].content + ticket_result = create_ticket.invoke({ + "summary": user_msg[:100], + "priority": "high" + }) + response = llm.invoke([ + SystemMessage(content="告知用户问题已升级，提供工单信息，安抚用户。"), + HumanMessage(content=f"工单创建结果：{ticket_result}") + ]) + return {"messages": [response], "final_answer": response.content} + + +# ============ 5. 路由逻辑 ============ +def route_by_classification(state: AgentState) -> Literal["simple_qa", "research", "escalation"]: + classification = state.get("classification", "") + if "simple_qa" in classification: + return "simple_qa" + elif "order_query" in classification: + return "research" + else: + return "escalation" + + +def should_continue_tools(state: AgentState) -> Literal["tools", "synthesize"]: + last_msg = state["messages"][-1] + if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: + return "tools" + return "synthesize" + + +# ============ 6. 构建图 ============ +# 流程： +# START → classify → route +# ├── simple_qa → END +# ├── research ⇄ tools → synthesize → END +# └── escalation → END + +workflow = StateGraph(AgentState) + +workflow.add_node("classify", classify_intent) +workflow.add_node("simple_qa", simple_qa_node) +workflow.add_node("research", research_node) +workflow.add_node("tools", tool_node) +workflow.add_node("synthesize", synthesize_node) +workflow.add_node("escalation", escalation_node) + +workflow.add_edge(START, "classify") +workflow.add_conditional_edges("classify", route_by_classification) +workflow.add_edge("simple_qa", END) +workflow.add_conditional_edges("research", should_continue_tools) +workflow.add_edge("tools", "research") +workflow.add_edge("synthesize", END) +workflow.add_edge("escalation", END) + +agent = workflow.compile() + + +# ============ 7. FastAPI 服务 ============ +app = FastAPI(title="LangGraph Agent Service") + +# 自动为 FastAPI HTTP 请求添加 tracing +FastAPIInstrumentor.instrument_app(app) + + +@app.post("/invoke") +async def invoke(request: Request): + """ + POST /invoke + Body: {"prompt": "你的问题", "user_id": "xxx", "session_id": "xxx"} + 返回: ndjson 流式响应，符合 ADK event 规范 + """ + body = await request.json() + prompt = body.get("prompt", "") + if not prompt: + return {"error": "prompt is required"} + + user_id = body.get("user_id", "default_user") + session_id = body.get("session_id", "default_session") + logger.info(f"user={user_id}, session={session_id}, prompt={prompt}") + + async def event_stream(): + # 只需一个顶层 span 标记请求入口，内部节点/LLM/工具的 span 由 LangSmith OTel 自动生成 + with tracer.start_as_current_span("agent.invoke") as span: + span.set_attribute("user_id", user_id) + span.set_attribute("session_id", session_id) + span.set_attribute("prompt", prompt[:200]) + + inputs = {"messages": [HumanMessage(content=prompt)]} + try: + async for chunk in agent.astream(inputs, stream_mode="updates"): + for node, state in chunk.items(): + logger.info(f"执行节点: {node}") + if "messages" in state and state["messages"]: + last_msg = state["messages"][-1] + content = last_msg.content if hasattr(last_msg, "content") else str(last_msg) + if content: + event_data = { + "content": { + "parts": [{"text": content}] + } + } + yield json.dumps(event_data) + "\n" + except Exception as e: + logger.error(f"Agent error: {e}") + span.set_status(trace.StatusCode.ERROR, str(e)) + span.record_exception(e) + event_data = {"content": {"parts": [{"text": f"处理出错: {str(e)}"}]}} + yield json.dumps(event_data) + "\n" + + return StreamingResponse(event_stream(), media_type="application/x-ndjson") + + +@app.get("/health") +async def health(): + return {"status": "ok"} + + +# ============ 8. 本地测试 ============ +# 三个 prompt 触发三条不同路径 +async def local_test(): + test_cases = [ + { + # 流程 1: classify → simple_qa → END + "prompt": "你们的退款政策是什么？", + "expected_flow": "classify → simple_qa → END", + }, + { + # 流程 2: classify → research → tools → research → synthesize → END + "prompt": "帮我查一下订单 ORD-20240315-001 到哪了", + "expected_flow": "classify → research ⇄ tools → synthesize → END", + }, + { + # 流程 3: classify → escalation → END + "prompt": "我已经投诉三次了，每次都说处理中但一直没结果，我要求退款并赔偿！", + "expected_flow": "classify → escalation → END", + }, + ] + + for i, case in enumerate(test_cases, 1): + print(f"\n{'='*60}") + print(f"测试 {i}: {case['prompt']}") + print(f"预期流程: {case['expected_flow']}") + print(f"{'='*60}") + + inputs = {"messages": [HumanMessage(content=case["prompt"])]} + nodes_visited = [] + async for chunk in agent.astream(inputs, stream_mode="updates"): + for node, state in chunk.items(): + nodes_visited.append(node) + if "messages" in state and state["messages"]: + last_msg = state["messages"][-1] + content = last_msg.content if hasattr(last_msg, "content") else "" + if content: + print(f" [{node}] {content[:120]}") + + print(f" 实际经过节点: {' → '.join(nodes_visited)}") + + +if __name__ == "__main__": + import sys + + if len(sys.argv) > 1 and sys.argv[1] == "test": + # python langchain_otel_agent.py test + import asyncio + asyncio.run(local_test()) + else: + # python langchain_otel_agent.py + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_guide.md b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_guide.md new file mode 100644 index 00000000..52292f13 --- /dev/null +++ b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/langchain_otel_guide.md @@ -0,0 +1,131 @@ +# LangSmith OTel 观测方案 — 对接 AgentKit APMPlus + +## 概述 + +本文档说明如何使用 LangChain 的 LangSmith OTel 集成，将 LangGraph Agent 的全链路 trace 数据上报到 AgentKit 平台的 APMPlus 观测后端。 + +对应示例代码：`langchain_otel_agent.py` + +## 架构原理 + +``` +LangGraph 节点 / LLM 调用 / 工具调用 + ↓ (LangSmith OTel 自动生成 span) +OpenTelemetry SDK (TracerProvider) + ↓ (OTLP gRPC + x-byteapm-appkey header) +AgentKit APMPlus 后端 +``` + +**核心思路**：利用 `langsmith.integrations.otel` 自动为所有 LangGraph 节点、LLM 调用、工具调用生成 span，再通过 OTLP exporter 将数据发送到 APMPlus，实现**零侵入式**的全链路追踪。 + +## 快速开始 + +### 1. 安装依赖 + +```bash +uv pip install -r requirements.txt +``` + +关键依赖： +- `langsmith[otel]` — LangSmith OTel 集成 +- `opentelemetry-sdk` — OTel 核心 +- `opentelemetry-exporter-otlp-proto-grpc` — OTLP gRPC 导出 +- `opentelemetry-instrumentation-fastapi` — FastAPI HTTP 自动打点 + +### 2. 配置环境变量 + +```bash +# APMPlus 观测后端（必需） +export OBSERVABILITY_OPENTELEMETRY_APMPLUS_ENDPOINT=apmplus-cn-beijing.ivolces.com:4317 +export OBSERVABILITY_OPENTELEMETRY_APMPLUS_API_KEY=your-api-key + +# LLM 配置 +export OPENAI_API_KEY=your-openai-key +export OPENAI_API_BASE=https://api.openai.com/v1 +export MODEL_NAME=gpt-4o-mini +``` + +### 3. 运行 + +```bash +# 启动 HTTP 服务（监听 8000 端口） +uv run langchain_otel_agent.py + +# 或本地测试（三条不同流程路径） +uv run langchain_otel_agent.py test +``` + +### 4. 请求示例 + +```bash +curl -N http://localhost:8000/invoke \ + -H 'Content-Type: application/json' \ + -d '{"prompt": "帮我查一下订单 ORD-20240315-001 到哪了"}' +``` + +## 代码接入说明 + +只需在应用启动时添加以下初始化代码，业务逻辑**零改动**： + +```python +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.resources import Resource +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter + +# 1. 配置 Resource（含 APMPlus 业务标识） +resource = Resource.create({ + "service.name": "langchain-agent", + "apmplus.business_carrier": "agentkit_runtime", +}) + +# 2. 配置 OTLP exporter（gRPC，带鉴权 header） +# 注意：gRPC metadata key 必须全小写 +exporter_headers = {"x-byteapm-appkey": os.getenv("OBSERVABILITY_OPENTELEMETRY_APMPLUS_API_KEY")} + +tracer_provider = TracerProvider(resource=resource) +tracer_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter( + endpoint=os.getenv("OBSERVABILITY_OPENTELEMETRY_APMPLUS_ENDPOINT"), + insecure=True, + headers=exporter_headers, + )) +) +trace.set_tracer_provider(tracer_provider) + +# 3. LangSmith OTel 自动追踪（挂到已有 TracerProvider） +from langsmith.integrations.otel import OtelSpanProcessor +tracer_provider.add_span_processor(OtelSpanProcessor()) +``` + +## 观测效果 + +在 APMPlus 后端可以看到： + +| Span 类型 | 说明 | +|-----------|------| +| HTTP span | FastAPI 自动注入的请求级 trace | +| agent.invoke | 每次请求的顶层 span（含 user_id、session_id、prompt） | +| LangGraph 节点 | classify → research → tools → synthesize 等节点自动追踪 | +| LLM 调用 | 每次 ChatOpenAI 调用的 input/output/tokens/latency | +| 工具调用 | 每次 tool 调用的参数和返回值 | + +Resource attributes 包含 `apmplus.business_carrier=agentkit_runtime`，便于在平台侧按业务载体筛选。 + +## Agent 多步骤流程 + +示例 Agent 使用 LangGraph 构建了多步骤工作流： + +``` +START → classify → route + ├── simple_qa → END (知识库直答) + ├── research ⇄ tools → synthesize → END (工具调用+综合) + └── escalation → END (升级工单) +``` + +三个测试 prompt 分别触发不同路径： + +1. `"你们的退款政策是什么？"` → classify → simple_qa → END +2. `"帮我查一下订单 ORD-20240315-001 到哪了"` → classify → research → tools → synthesize → END +3. `"我已经投诉三次了...我要求退款并赔偿！"` → classify → escalation → END diff --git a/01-tutorials/workshop/session1/langchain_agent_deploy_sample/requirements.txt b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/requirements.txt index 20cea463..fcc5edcc 100644 --- a/01-tutorials/workshop/session1/langchain_agent_deploy_sample/requirements.txt +++ b/01-tutorials/workshop/session1/langchain_agent_deploy_sample/requirements.txt @@ -1,3 +1,11 @@ langchain>=1.1.3 langchain-openai>=1.1.3 +langgraph>=0.4.0 python-dotenv>=1.2.1 +fastapi>=0.115.0 +uvicorn>=0.34.0 +langsmith[otel]>=0.3.0 +opentelemetry-api>=1.28.0 +opentelemetry-sdk>=1.28.0 +opentelemetry-exporter-otlp-proto-grpc>=1.28.0 +opentelemetry-instrumentation-fastapi>=0.49b0