聊到LLM应用的可观测性绕不开一个老生常谈的问题单次调用追踪好做可一旦到了多轮对话链路就乱成了一团麻。用户上一句还在问天气下一句已经开始聊行程推荐如果只盯着单个trace你根本看不出模型在哪个环节开始失忆也不知道是哪一轮的上下文把token预算拖爆了。我在项目里被这个问题折腾过很多次后来切到Opik的Threads功能才总算理顺。Opik是Comet那一套体系里开源出来的LLM可观测性平台支持traces、spans、数据集管理和在线评估而Threads就是它对多轮会话场景做追踪的设计方案。这篇文章我会从底层逻辑讲到落地操作把多轮对话的追踪链路整个拆开配合真实的代码和踩坑记录做成一份可以直接照着用的实操指南。无论你是在调聊天机器人、客服Agent还是任何需要跨多轮维持状态的LLM应用这篇内容都适用。1. Threads 到底是什么一次搞懂多轮对话追踪的核心逻辑先说一个常见的误解。很多人以为Threads是Opik里一个单独的追踪模块要用就要调用一套新API。实际上完全不是这么回事。Threads本质上是一套基于关联维度的组织方式它的核心不是新协议而是一个贯穿多轮调用的标识——thread_id。理解这一点后面所有操作都会顺理成章。1.1 一个trace只能看单次Threads解决的是连续性问题在Opik里trace对应的是你应用对外提供一次完整服务的调用过程。比如用户问了一句你好你的服务走完检索、调用模型、返回答案这整条路径就是一个trace。它内部可以有多个span分别记录工具执行、模型请求、解析耗时等细节。这个设计对单轮请求来说非常清晰问题在于它无法表达轮次之间的关联。多轮对话有几个特征是单次trace天然无法覆盖的上下文会在多轮之间传递前一轮的输出可能成为后一轮的系统提示词内容。某些问题必须回溯前面若干轮才能定位比如我刚说的那个地址你记得到吗。token消耗和质量问题是跨轮叠加的只看一次调用分析不出来。Threads做的就是给这些原本分散的trace贴上一个共同的标签。你在同一场对话中创建的所有trace只要携带同一个thread_id进入Opik界面后就能按这个ID聚合成一个可展开的时间线。你可以从上到下完整看到这场会话发生了什么而不是去海量trace记录里手动筛选。1.2 关联方式thread_id 不是玩具它是一条贯穿会话的线索具体来说thread_id是一个普通的字符串标识。它可以是雪花ID、UUID、数据库自增ID的字符串形式甚至是你业务里的session_id。Opik不关心它怎么生成只关心它在同一会话里是否一致。通常我们把thread_id设置在trace上所有的子span会自动继承这个关联关系。也就是说你只需要在创建最外层trace时指定thread_id里面套了多少层工具调用、多少次模型请求它们都会被归到同一个thread下。这里有个对新手很友好的一点Opik允许同一个thread_id被多个trace使用。换句话说用户每发一轮新消息你不需要新建一个thread只需用相同的thread_id再创建一个trace就够了。Opik内部会把相同thread_id的trace按时间顺序排列在一起形成完整会话流。这也是Threads和传统链路追踪比如OpenTelemetry的trace最大的区别之一——后者通常要求每个请求一个唯一trace ID而Opik的Threads天然是一对多的关系模型。1.3 什么时候该用 Threads什么时候不该用不是所有场景都适合用Threads。如果只是做离线评测或者单轮问答测试强制引入thread_id反而增加无用开销。我的判断标准很简单只要一次用户操作可能依赖之前的状态就需要Threads。适合用Threads的典型场景聊天机器人用户会连续发多条消息。客服Agent客户和坐席之间来回沟通。多轮工具调用比如AI在完成一个任务时需要一步步确认信息。有状态Agent比如角色扮演、教育辅导类的对话产品。不适合用Threads的典型场景纯单轮问答每次对话互相独立。离线批量评测只需要对单条样本打分。非对话类的后台任务比如定时跑批处理。判断完场景之后接下来就可以动手接入了。2. 快速上手十分钟跑通第一个 Threads 追踪在写代码之前先把环境准备好。Opik支持云服务需要注册账号拿API Key也支持本地自托管。对绝大多数团队来说本地起一套Docker镜像是最省事的方式尤其适合数据敏感或需要离线开发的项目。2.1 安装与初始化安装就一条命令pip install opik初始化客户端有两种方式。官方文档里最常见的是通过环境变量或opik.configure来设置全局参数然后再创建Opik对象。如果你用的是本地自托管部署host地址填你自己服务暴露的端口就行。import opik opik.configure( api_keyyour_api_key, # 本地自托管可以随便填 hosthttp://localhost:5173 # 自托管服务地址 ) client opik.Opik()这里提醒一句opik.configure并不是必须的如果你直接Opik()SDK会去读环境变量OPIK_API_KEY和OPIK_HOST。我习惯在开发环境里用dotenv管理这些变量然后把configure这步省掉代码看起来更干净。生产环境则建议把这些配置放到部署平台的环境变量里不要写在代码中。2.2 单轮链路先搭好先别急着谈多轮我们先把单轮trace跑通。以此为基础后面加thread_id只是改一行的事。opik.track def generate_answer(question: str): # 模拟一次LLM调用 response call_llm(question) return response with client.trace( namesingle-turn-chat, metadata{user_id: u_123}, ) as trace: answer generate_answer(你好介绍一下你自己) trace.log_output({answer: answer})这里的opik.track装饰器会自动为函数创建一个span并把输入输出记录下来。如果函数内部还有对另一个模型API的调用你也可以在函数内部再包一个opik.track标注的辅助函数这样一层层嵌套下去最终形成一个完整span树。先确保这些基础链路在UI里能看到再去加多轮关联。2.3 把多轮对话串成 Thread多轮对话的核心改造点就一个在每一次用户请求创建trace时把同一个thread_id传进去。来看一个实际可运行的示例import uuid import opik from opik import track client opik.Opik() # 模拟一个chat session def chat_handler(): session_id str(uuid.uuid4()) # 每个会话生成一个唯一ID conversation [ {role: user, content: 我想规划一次杭州三天两夜的行程}, {role: assistant, content: 好的我先帮你整理几个核心景点}, {role: user, content: 主要想去西湖和灵隐寺帮我安排一下时间}, {role: assistant, content: 建议第一天去西湖第二天上午灵隐寺下午可以逛逛周边的茶园}, {role: user, content: 那住宿选在哪里比较方便}, {role: assistant, content: 推荐住在湖滨银泰附近去西湖和灵隐寺都比较方便}, ] for i in range(0, len(conversation), 2): user_msg conversation[i][content] with client.trace( namechat-turn, thread_idsession_id, metadata{turn: i // 2 1} ) as trace: answer generate_answer(user_msg) trace.log_output({answer: answer}) track def generate_answer(question: str): # 你的真实LLM调用逻辑 return f这是对{question}的回答上面的代码里thread_id在整个会话循环中保持不变每一轮用户输入都会开启一个新的trace但它们都会被归到同一个Thread下。UI里的展示效果就是你能看到这场杭州行程规划对话的完整现场回放每一轮LLM的输入输出都带上了上下文顺序。提示thread_id一般不要直接用用户ID。原因后面会细说。先用独立生成的会话ID比如UUID后续就算需要关联到具体用户也可以在metadata里同时存一个user_id这样既能在Thread中查看会话又能通过user_id去筛业务数据。3. 进阶技巧全局配置、跨进程关联与 UI 实战基础跑通之后真正复杂的往往是实际业务场景中的各种约束。比如你在FastAPI里给每个请求都起了一个新协程thread_id要怎么传进去比如前后端分离浏览器端的会话ID和后端LLM服务的thread_id怎么对齐这些坑一个比一个现实。3.1 全局 thread_id 的坑与解Opik提供了一个全局配置方式可以直接设置默认的thread_id这样所有未显式指定thread_id的trace都会自动归属到这个线程下。import opik opik.configure(thread_idmy-global-thread-id)用的时候千万注意opik.configure内部默认会把值定义成全局线程变量实现上类似contextvars。如果你在异步代码里随手调用它很可能出现不同请求互相覆盖thread_id的情况然后你会发现所有trace都串到了同一个Thread上好几天数据都是乱的。正确做法是在线程/协程开头设置thread_id用完及时清理或使用上下文管理器。尽量在创建的trace上显式传thread_id让全局配置只做兜底。异步场景不要滥用configure(thread_id...)优先用contextvars.copy_context()配合SDK追踪或者在HTTP请求中间件层统一注入。如果你用的框架自带上下文管理比如FastAPI的request scope里用contextvars那可以自己写一个非常轻的依赖注入把thread_id绑定到请求级context。这样可以保证不同用户、不同会话之间互不干扰。3.2 跨进程/跨服务传递 thread_id很多LLM应用不是单机跑一个Python文件而是拆成了API网关、业务服务、模型网关、RAG服务等多个模块。这时候thread_id需要透传不能只在最上层设置完就指望底层所有组件都自动带上。我惯用的方案是通过消息头或上下文对象传递。比如在HTTP服务里from fastapi import Request, Header app.post(/chat) async def chat_endpoint(request: Request, x_thread_id: str Header(default)): thread_id x_thread_id or str(uuid.uuid4()) # 后续所有调用都显式传这个thread_id ...如果是走消息队列或事件总线可以把thread_id塞进事件体里消费者取出后通过opik.track(thread_id...)或者client.span(thread_id...)挂在trace上。这里有一个容易忽略的细节如果你在消费端创建了新的trace记得把thread_id也传给消费端的trace否则生产端和消费端会各自形成两个Thread。还有一个更简单的思路直接全局设置opik.configure(thread_idthread_id)然后在处理完这条消息后把它重置为None。这么做的问题在于消息处理往往是高并发的一旦并发上来就会互相污染。所以我的建议始终是显式传参优先于全局配置。3.3 UI 分析与过滤技巧Opik界面中Thread信息会在Trace详情页右侧展示你可以看到该Thread包含的所有traces按时间顺序排列。调试多轮对话时我一般这么操作在Trace列表中按thread_id过滤。Opik默认支持直接在过滤条件里输入thread_id做精确匹配。进入某个trace的详情页后看它归属的Thread整体时间线。关注每一轮trace的token消耗和时延变化。如果第三轮开始时延从500ms跳到2s就要去检查是不是上下文过长导致模型需要处理更多token。对比不同用户对同一thread的反馈分数用来发现特定会话的长尾问题。在UI里还有一个比较实用的查看方式以Thread为维度去聚合token和成本。虽然没有现成的超强报表但你可以利用metadata里的字段做分组或者把Opik数据导出后在BI工具里二次分析。我也见过有团队直接把Opik的API数据拉下来用pandas做会话级质量分析效果很直观。4. 实战案例用 Threads 分析一个客服机器人的完整会话光讲概念不够直观我们用一个完整的客服机器人场景把Threads从前到后串一遍。这个案例会尽量贴近真实业务包含会话建模、代码落地、UI分析三个部分。4.1 场景建模假设我们在做一个电商平台的售后客服机器人。用户会进来问物流、退款、换货等问题。典型的流程是用户说明问题Agent调用订单查询工具再结合问题类型走不同的答复逻辑。多轮对话可能长这样第1轮用户问我的订单什么时候发货第2轮Agent查询订单返回预计发货时间第3轮用户继续问那能不能改一下收货地址第4轮Agent查订单状态确认是否允许修改再执行修改操作这个场景下Threads的用途非常明显。如果只记录单个trace你根本不知道第4轮的改地址操作其实是基于第1轮那个订单上下文的。而把所有trace挂在同一个thread_id下整个流程一目了然。4.2 关键代码实现我们先封装一个模拟的订单服务函数让它真的调用一个工具并返回结果。整个过程会模拟真实的LLM工具调用链路。import uuid import json import opik from opik import track client opik.Opik() track def query_order(order_id: str): 模拟订单查询工具 # 假设这里调用了真实的订单系统API return { order_id: order_id, status: shipped, eta: 2025-04-20, shipping_address: 杭州市西湖区xxx路xx号 } track def update_address(order_id: str, new_address: str): 模拟修改收货地址工具 return { order_id: order_id, new_address: new_address, updated: True } track def agent_reply(user_message: str, context: dict): 模拟LLM根据上下文生成回复 # 真实项目中这里是调用大模型 return f根据查询结果{json.dumps(context, ensure_asciiFalse)} def process_turn(thread_id: str, user_message: str, turn_number: int): with client.trace( namecustomer-service-turn, thread_idthread_id, metadata{turn_number: turn_number, channel: web} ) as trace: trace.log_input({user_message: user_message}) # 模拟简单的意图判断 if 发货 in user_message or 物流 in user_message: order_info query_order(order_idORD123456) reply agent_reply(user_message, order_info) elif 改地址 in user_message or 收货地址 in user_message: order_info query_order(order_idORD123456) update_result update_address(order_idORD123456, new_address上海市浦东新区xx路xx号) reply agent_reply(user_message, {**order_info, **update_result}) else: reply agent_reply(user_message, {hint: 无法处理}) trace.log_output({reply: reply}) return reply if __name__ __main__: session_id str(uuid.uuid4()) turns [ 我的订单什么时候发货, 那能不能改一下收货地址, 帮我改成默认地址, ] for idx, msg in enumerate(turns, start1): process_turn(thread_idsession_id, user_messagemsg, turn_numberidx)上面这段代码里需要注意的点每次process_turn都会创建一个新的trace但它们共享同一个session_id。query_order和update_address加了track装饰器因此它们会成为当前trace底下的span自动归入同一thread。我把turn_number写进了metadata方便后续在UI或导出数据里按轮次排序、聚合。跑完这段代码打开Opik UI搜索session_id对应的thread你能看到完整的3个trace每个trace内部还有多层span结构。这种建模方式对小型项目刚刚好不重也不轻。4.3 在UI中分析会话质量代码跑完之后分析才是重头戏。我在实际项目里会重点看这几个维度时延分布。多轮对话中每一轮的耗时是独立的如果第1轮正常第2轮开始突然变慢大概率是上下文变长导致模型推理变慢也可能是检索工具在第二轮的召回量暴增。在UI里按trace查看时延分布能很快锁定是哪一种。一致性。如果同一场会话里用户第1轮说想改地址第2轮Agent却还在回答预计4月20日发货说明模型上下文里对于用户意图的追踪出了问题。在Threads 时间线里可以把前后两轮的输出连起来看这种bug一眼就能发现。成本估算。每轮trace都有token记录把同一个thread下的token数加起来就是这场对话的总消耗。这对于控制客服机器人这类高频交互场景的成本非常有价值。反馈关联。Opik支持给批量trace打得分你可以在UI里对某个trace标记好或坏。如果某个用户最终选择转人工或者点了不满意你可以给该轮trace打上低分再反向看这个thread里的前几轮到底出了什么问题。5. 常见问题与排查技巧实录这部分是实际操作中容易踩的坑我按照问题出现的频率整理一下。5.1 trace 没有按 thread 聚在一起现象明明在trace上设置了thread_idUI里却搜不到完整的会话。排查方向先确认不同轮次的thread_id字符串完全一致。别小看这个问题多一个空格、大小写不一致都会导致关联失败。我在项目里就遇到过把uuid4().hex和uuid4().int混用的情况两种方式生成的结果格式不同硬是排查了很久。确认服务端版本和SDK版本兼容。Opik迭代速度比较快有些老版本对于thread_id的字段名解析存在差异。最简单的方法是统一升级到当前最新稳定版。检查是不是设了全局thread_id又被局部覆盖了。如果你在某个span上设置了一个不同的thread_id它可能会覆盖trace级的配置导致挂在别的线程下。5.2 thread_id 在异步场景丢失现象使用FastAPI异步协程或Celery异步任务时thread_id没有传递到子任务里。原因opik.track装饰器对async函数的支持没有问题但如果你在协程里手动创建了新task或者用asyncio.gather并发执行多个函数这些函数运行时各自是独立的context无法直接读取到父协程里设置的thread_id配置。解决方案import asyncio from contextvars import copy_context async def task_worker(question: str, thread_id: str): with client.trace(nameasync-task, thread_idthread_id) as trace: trace.log_input({question: question}) # do something return done async def main(): thread_id session-123 ctx copy_context() # 通过copy_context把thread_id上下文传到子协程 results await asyncio.gather( *[asyncio.create_task(ctx.run(lambda qt: task_worker(q, thread_id))) for t in range(3)] )简而言之别在异步并发场景里依赖全局thread_id每个task启动时显式把thread_id作为参数传进去最稳妥。5.3 同一线程被多次创建这个问题主要出现在前后端联调时。前端页面每次刷新都会重新生成一个session_id如果后端是用session_id来当thread_id那用户刷新一次页面就会产生一个新Thread历史会话就断了。我的处理方式前端持久化session_id到localStorage或后端会话表里每次刷新后优先从持久化存储读取而不是重新生成。后端增加一个会话恢复接口用户重新连接时传入旧session_id服务端校验后继续沿用。不要把thread_id等同于HTTP session。thread_id更偏业务语义可以比session的粒度更细比如一个用户一次提问代表一个turn而一个客服任务会话包含多个turn这个任务会话才叫thread。5.4 性能开销问题很多人在上线前会担心加追踪会不会带来明显性能损耗Opik的SDK是异步上报的本地会将trace数据先做缓存再批量发送到服务端。正常情况下对业务接口延迟影响很小。但有几个点要注意不要在高频路径上同步调用client.flush()这会导致当前线程阻塞等待网络I/O。如果上报的服务端响应特别慢SDK内部缓存队列可能会积压内存占用上升。可以把上报间隔调大或者在低峰期做批量导出。大部分情况下不用开debug日志生产环境把日志级别调到WARNING以上避免大量trace上报日志刷屏。我见过一个项目因为自托管Opik服务器配的机器比较小日志一多直接卡死最终通过把OPIK_LOG_LEVEL调成ERROR以及限制SDK上报并发数才缓解。6. 把 Threads 用进你的日常调试流当工具链稳定之后Threads带来的最直观改变是我现在看问题单真的只剩望闻问切了。拿到一个用户反馈机器人回答很奇怪我先根据反馈里的会话时间去找到对应thread把时间线展开看几轮再对照metadata里的channel、版本号等信息基本就能定位是意图识别跑偏还是上下文传递丢失还是Prompt写得不合适。整个过程不会超过五分钟解决之后还能在trace上打个低分标记作为bad case积累。对于还在用单次trace排查多轮问题的团队我真的建议尽快把thread_id体系接进去。改动其实很小就是每次创建trace多传一个参数但换来的是会话级视野。它会慢慢改变你调试和优化LLM应用的方式让你从看一棵树变成看一整片森林。等上了这套体系之后你再回去看那些时好时坏的对话case多半会感叹一句原来问题一直就在第2轮的上下文里藏着。
