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.mdPython 版本 3.10 3.13安装依赖pip install agentops langchain langchain_openai示例目录中的 requirements.txt 仅固定了两个 LangChain 侧依赖langchain与langchain-openaiagentops本身按 README 指引单独安装。此外完整示例代码用到了python-dotenv来加载.env中的 API Key在 notebook 中通过%pip install python-dotenv安装见 langchain_examples.ipynb。本目录提供两种等价的示例形态脚本形式langchain_examples.pyNotebook 形式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], modelgpt-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(agentagent, toolstools) agent_executor.invoke({input: What comedies are playing?}, config{callback: [agentops_handler]}) # 7. 编程化验证 Span 是否成功上报 import agentops agentops.validate_trace_spans(trace_contextNone)三个挂载点callbacks 应该传给谁从示例代码可以总结出 handler 的三个标准挂载位置这是数据能否完整上报的关键LLM 实例ChatOpenAI(callbacks[agentops_handler], ...)捕获模型调用on_chat_model_start/on_llm_end等回调每个工具t.callbacks [agentops_handler]捕获工具调用on_tool_start/on_tool_end执行入口agent_executor.invoke(..., config{callback: [agentops_handler]})把 handler 传入 AgentExecutor 的运行配置保证整条执行链路上的事件都能被拦截。langchain/callbacks 集成文档 的 Troubleshooting 一节也印证了这一点如果 Dashboard 看不到数据检查项之一就是 “确保你把 handler 传给了所有相关组件Ensure youre passing the handler to all relevant components”。Handler 构造参数与自动初始化LangchainCallbackHandler的构造函数接受三个参数见 callback.py#L37-L54参数类型默认值说明api_keyOptional[str]NoneAgentOps API Key不传时回退到AGENTOPS_API_KEY环境变量tagsOptional[List[str]]None内部转空列表附加到 Session 的标签便于在 Dashboard 中筛选会话auto_sessionboolTrue是否自动创建 Session 根 Span 并初始化 AgentOps当auto_sessionTrue时构造函数会调用_initialize_agentops()其内部逻辑callback.py#L56-L93为若全局 tracer 尚未初始化则以固定参数调用agentops.init(auto_start_sessionFalse, instrument_llm_callsTrue)——即让 callback handler 自己接管 Session 的创建同时开启 LLM 调用插桩如果构造时传了api_key会一并传入init。创建一个名为session.SESSION的根 Span写入属性session.tags、agentops.operation.namesession、span.kindSESSION等。通过attach(set_span_in_context(self.session_span))将 Session Span 挂到当前 OpenTelemetry 上下文后续所有子 Span 都会以此为父。也就是说示例代码中“创建 handler 之后 session 就被自动记录”并非黑箱而是上述初始化链的结果。Span 层级run_id 驱动的父/子关系维护LangChain 的每次运行run都会携带run_id与parent_run_idhandler 正是靠这对 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” 的容错设计。异步版本委托给同步 HandlerAsyncLangchainCallbackHandlercallback.py#L683-L705继承AsyncCallbackHandler但内部实现非常简洁构造时创建一个LangchainCallbackHandler实例所有async def on_xxx方法都只是转发到同步 handler 的对应方法。因此异步场景如ainvoke与同步场景享有完全一致的 Span 语义与清理行为。它同样通过init.py 对外导出。支持的回调方法与 Span 属性LangchainCallbackHandler实现了 LangChain 回调体系的完整方法集映射关系与记录的属性如下来自 langchain/callbacks 集成文档回调方法说明Span Kind记录属性on_llm_startLLM 调用开始llm模型、prompts、参数on_llm_endLLM 调用结束llm补全文本、token 用量on_llm_new_token流式 token 到达无token 计数、末位 tokenon_llm_errorLLM 调用异常llm错误详情on_chat_model_startChat 模型调用开始llm模型、消息、参数on_chain_startChain 开始taskChain 类型、输入on_chain_endChain 结束task输出on_chain_errorChain 执行异常task错误详情on_tool_start工具调用开始tool工具名、输入on_tool_end工具调用结束tool输出on_tool_error工具执行异常tool错误详情on_agent_actionAgent 采取动作agent工具、输入、日志on_agent_finishAgent 完成任务agent输出、日志on_text任意文本事件text文本内容源码中可以看到几个值得注意的实现细节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_TOKENScallback.py#L479-L503。错误属性三类on_xxx_error回调都会写入errorTrue、错误类型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_contextNone)做端到端验证。该函数定义在 validation.py#L209-L217关键参数如下参数默认值说明trace_idNone直接指定要校验的 trace IDtrace_contextNonestart_trace返回的 TraceContext与trace_id二选一max_retries10等待 Span 出现的最大重试次数retry_delay1.0每次重试间隔秒check_llmTrue是否专门检查 LLM Spanmin_spans1期望的最少 Span 数量api_keyNone可选 API Key缺省时使用环境变量当两者都未提供时函数会尝试从当前 OpenTelemetry 上下文的 Span 中提取trace_idvalidation.py#L238-L252拿不到 JWT token 时会返回validation_skipped: True并说明原因而不是直接抛错。校验失败则抛出ValidationError——示例代码正是捕获该异常并打印错误信息的。工作原理小结与故障排查从源码结构看handler 的完整工作链路为初始化时创建 Session 根 Spanauto_sessionTrue时拦截 LangChain 各类回调事件按run_id/parent_run_id创建带属性的 Span 并维护父/子关系操作完成时自动关闭 Span 并清理上下文。如果按上述方式接线后 Dashboard 仍然没有数据可以按 集成文档 的 Troubleshooting 逐项排查确认 API Key 配置正确环境变量AGENTOPS_API_KEY或构造参数api_key确认 handler 已传给所有相关组件LLM、工具、执行入口三处挂载点确认所有操作正常结束/关闭若程序异常退出依赖__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.mdSpan 上报校验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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考