企业官网建设流程全解析

AgentOps × LangChain：用 Callback Handler 实现对 LLM 应用与 Agent 的完整追踪

【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops

本篇技术指南围绕 AgentOps 仓库中的 LangChain 集成示例（examples/langchain/）展开，讲解如何通过LangchainCallbackHandler将 LangChain 的 LLM 调用、Chain、工具调用和 Agent 行为自动记录为 Session 与 Span。读完本文，你将掌握环境准备与依赖安装、完整的工具调用 Agent 示例代码，以及该回调处理器在 OpenTelemetry 层面的底层实现原理（父/子 Span 层级、流式 Token 计数、错误捕获），并学会用validate_trace_spans验证遥测数据是否成功上报。

环境要求与依赖安装

LangChain 示例的运行环境约束与安装步骤如下（源自 examples/langchain/README.md）：

• Python 版本：>= 3.10 < 3.13
• 安装依赖：

pip install agentops langchain langchain_openai

示例目录中的 requirements.txt 仅固定了两个 LangChain 侧依赖（langchain与langchain-openai），agentops本身按 README 指引单独安装。此外，完整示例代码用到了python-dotenv来加载.env中的 API Key，在 notebook 中通过%pip install python-dotenv安装（见 langchain_examples.ipynb）。

本目录提供两种等价的示例形态：

• 脚本形式：langchain_examples.py
• Notebook 形式：langchain_examples.ipynb

两者演示的是同一件事：一个带工具调用的 LangChain Agent，如何被 AgentOps 自动插桩并记录到 Dashboard。

核心集成思路：一个 Handler 承担全部插桩

AgentOps 对 LangChain 的集成基于 LangChain 原生的回调机制：LangchainCallbackHandler继承自langchain_core.callbacks.base.BaseCallbackHandler，实现文件位于 callback.py。它会在初始化时自动完成 AgentOps 的初始化并创建 Session 根 Span，随后拦截 LangChain 运行时触发的各类回调事件，为每次 LLM 调用、Chain 执行、工具调用和 Agent 动作创建对应的 Span。

接入的完整流程在 langchain_examples.py 中演示，以下代码为完整可运行的示例（省略 API Key 占位说明）：

import os from langchain_openai import ChatOpenAI from langchain.agents import tool, AgentExecutor, create_openai_tools_agent from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate # AgentOps 唯一的“侵入点”：导入这个特殊的 Callback Handler from agentops.integration.callbacks.langchain import ( LangchainCallbackHandler as AgentOpsLangchainCallbackHandler, ) # 1. 配置 API Key（.env 中的 AGENTOPS_API_KEY / OPENAI_API_KEY，或内联设置） load_dotenv() os.environ["AGENTOPS_API_KEY"] = os.getenv("AGENTOPS_API_KEY", "your_api_key_here") os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "your_openai_api_key_here") # 2. 创建 Handler 实例，传入 tags 便于在 Dashboard 中检索该会话 # Handler 初始化后，一个 session 会被自动创建 agentops_handler = AgentOpsLangchainCallbackHandler(tags=["Langchain Example", "agentops-example"]) # 3. 将 handler 挂到 LLM 的 callbacks 上 llm = ChatOpenAI(callbacks=[agentops_handler], model="gpt-3.5-turbo") llm.callbacks = [agentops_handler] prompt = ChatPromptTemplate.from_messages( [ ("system", "You are a helpful assistant. Respond only in Spanish."), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) # 4. 定义工具：工具使用同样会被记录 @tool def find_movie(genre: str) -> str: """Find available movies""" if genre == "drama": return "Dune 2" else: return "Pineapple Express" tools = [find_movie] # 5. 为每个工具补充 callback handler for t in tools: t.callbacks = [agentops_handler] llm_with_tools = llm.bind_tools([find_movie]) # 6. 创建 Agent 并执行，所有动作都会记录到 Dashboard agent = create_openai_tools_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools) agent_executor.invoke({"input": "What comedies are playing?"}, config={"callback": [agentops_handler]}) # 7. 编程化验证 Span 是否成功上报 import agentops agentops.validate_trace_spans(trace_context=None)

三个挂载点：callbacks 应该传给谁

从示例代码可以总结出 handler 的三个标准挂载位置，这是数据能否完整上报的关键：

1. LLM 实例：ChatOpenAI(callbacks=[agentops_handler], ...)，捕获模型调用（on_chat_model_start/on_llm_end等回调）；
2. 每个工具：t.callbacks = [agentops_handler]，捕获工具调用（on_tool_start/on_tool_end）；
3. 执行入口：agent_executor.invoke(..., config={"callback": [agentops_handler]})，把 handler 传入 AgentExecutor 的运行配置，保证整条执行链路上的事件都能被拦截。

langchain/callbacks 集成文档 的 Troubleshooting 一节也印证了这一点：如果 Dashboard 看不到数据，检查项之一就是 “确保你把 handler 传给了所有相关组件（Ensure you're passing the handler to all relevant components）”。

Handler 构造参数与自动初始化

LangchainCallbackHandler的构造函数接受三个参数（见 callback.py#L37-L54）：

当auto_session=True时，构造函数会调用_initialize_agentops()，其内部逻辑（callback.py#L56-L93）为：

1. 若全局 tracer 尚未初始化，则以固定参数调用agentops.init(auto_start_session=False, instrument_llm_calls=True)——即让 callback handler 自己接管 Session 的创建，同时开启 LLM 调用插桩；如果构造时传了api_key，会一并传入init。
2. 创建一个名为session.SESSION的根 Span，写入属性session.tags、agentops.operation.name=session、span.kind=SESSION等。
3. 通过attach(set_span_in_context(self.session_span))将 Session Span 挂到当前 OpenTelemetry 上下文，后续所有子 Span 都会以此为父。

也就是说，示例代码中“创建 handler 之后 session 就被自动记录”并非黑箱，而是上述初始化链的结果。

Span 层级：run_id 驱动的父/子关系维护

LangChain 的每次运行（run）都会携带run_id与parent_run_id，handler 正是靠这对 ID 还原出调用树：

• _create_span()（callback.py#L95-L156）：把 Span 存入self.active_spans[run_id]字典；如果parent_run_id命中某个活跃 Span，就以该 Span 作为父上下文创建子 Span；否则直接挂到 Session 根 Span 下。同时用attach/set_span_in_context维护上下文令牌，存入self.context_tokens[run_id]。
• _end_span()（callback.py#L158-L183）：从active_spans弹出 Span、detach对应上下文令牌、调用span.end()，并清理流式 Token 计数器。若找不到对应 Span 会记录 warning 而不是抛异常，保证插桩失败不会打断业务代码。
• 处理器在__del__中做兜底清理：结束所有未关闭的 Span、结束 Session Span 并 detach 其上下文令牌（callback.py#L454-L477）。这对应集成文档中 “自动清理并在操作完成时结束 Span” 的容错设计。

异步版本：委托给同步 Handler

AsyncLangchainCallbackHandler（callback.py#L683-L705）继承AsyncCallbackHandler，但内部实现非常简洁：构造时创建一个LangchainCallbackHandler实例，所有async def on_xxx方法都只是转发到同步 handler 的对应方法。因此异步场景（如ainvoke）与同步场景享有完全一致的 Span 语义与清理行为。它同样通过init.py 对外导出。

支持的回调方法与 Span 属性

LangchainCallbackHandler实现了 LangChain 回调体系的完整方法集，映射关系与记录的属性如下（来自 langchain/callbacks 集成文档）：

源码中可以看到几个值得注意的实现细节：

• LLM 参数捕获：on_llm_start与on_chat_model_start会从serialized["kwargs"]中提取temperature、max_tokens、top_p写入 Span 属性（callback.py#L204-L211），使成本与行为分析能关联到具体生成参数。
• Token 用量：on_llm_end从response.llm_output["token_usage"]中分别提取prompt_tokens、completion_tokens、total_tokens写入对应 Span 属性（callback.py#L253-L272）。
• 流式 Token 计数策略：on_llm_new_token故意不对每个 token 都设置属性——源码注释明确说明逐 token 写属性既低效又可能触发 “setting attribute on ended span” 错误；它只把计数累加到self.token_counts[run_id]，在on_llm_end时一次性写入LLM_USAGE_STREAMING_TOKENS（callback.py#L479-L503）。
• 错误属性：三类on_xxx_error回调都会写入error=True、错误类型（error.__class__.__name__）、错误消息，然后正常结束 Span，保证失败调用在追踪树中依然可见。
• Chain 类型推断：on_chain_start会根据 chain 名称中含sequential/llm/router字样推断 Chain 的子类型属性（callback.py#L303-L309）。

模型名提取逻辑

LLM Span 上的模型名由 utils.py 中的get_model_info()提取。它按优先级依次尝试：serialized["id"]列表的末位元素、serialized["model_name"]字符串、带/分隔的id（取split("/", 1)后的后半段）、以及serialized["kwargs"]中的model_name/model字段；任何一级解析失败都会兜底为"unknown"。这解释了为什么在 Dashboard 里即使不同 LangChain 版本的序列化格式有差异，模型名一般也能正确显示。

验证 Span 是否成功上报

示例脚本的最后一段调用agentops.validate_trace_spans(trace_context=None)做端到端验证。该函数定义在 validation.py#L209-L217，关键参数如下：

当两者都未提供时，函数会尝试从当前 OpenTelemetry 上下文的 Span 中提取trace_id（validation.py#L238-L252）；拿不到 JWT token 时会返回validation_skipped: True并说明原因，而不是直接抛错。校验失败则抛出ValidationError——示例代码正是捕获该异常并打印错误信息的。

工作原理小结与故障排查

从源码结构看，handler 的完整工作链路为：

1. 初始化时创建 Session 根 Span（auto_session=True时）；
2. 拦截 LangChain 各类回调事件；
3. 按run_id/parent_run_id创建带属性的 Span 并维护父/子关系；
4. 操作完成时自动关闭 Span 并清理上下文。

如果按上述方式接线后 Dashboard 仍然没有数据，可以按 集成文档 的 Troubleshooting 逐项排查：

1. 确认 API Key 配置正确（环境变量AGENTOPS_API_KEY或构造参数api_key）；
2. 确认 handler 已传给所有相关组件（LLM、工具、执行入口三处挂载点）；
3. 确认所有操作正常结束/关闭（若程序异常退出，依赖__del__兜底，但实时性会受限）。

相关源码与文档路径汇总，便于继续深入：

• 示例脚本：examples/langchain/langchain_examples.py
• 示例依赖：examples/langchain/requirements.txt
• 回调处理器实现：agentops/integration/callbacks/langchain/callback.py
• 模型信息提取：agentops/integration/callbacks/langchain/utils.py
• 集成行为说明：agentops/integration/callbacks/langchain/README.md
• Span 上报校验：agentops/validation.py

【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops