网硕互联帮助中心如果你正在开发聊天机器人或对话式 AI 应用，一定会遇到一个棘手的问题：如何追踪用户和 AI 之间的完整对话？单次问答的追踪已经够复杂了，多轮对话更是让上下文管理、质量评估和问题排查变得难上加难。Opik 提供了一套优雅的解决方案——Threads。它能把相关的追踪记录分组到一起，形成一条完整的对话线程，让你轻松回顾多轮交互、追踪用户会话，并对整个对话进行评分。
这篇文章会从零开始，带你了解 Opik 中 Threads 的概念、如何记录对话、Thread ID 的最佳实践，以及如何审查和评分。无论你用的是 TypeScript 还是 Python，无论你依赖 LangGraph、ADK 还是 OpenAI Agents，都能找到对应的集成方式。
为什么需要 Threads？
在传统的单次请求追踪中，每个 trace 代表一次独立的交互。但聊天场景不一样。用户问“什么是机器学习？”，AI 回答后，用户接着问“能举个例子吗？”，这两轮对话在逻辑上是连续的，上下文是累积的。如果只把它们看作两个孤立的 trace，你就无法还原对话的全貌，也无法分析多轮交互中的连贯性和用户满意度。
Opik 的 Threads 就是为解决这个问题而生的。Thread 是一组相关 trace 的集合，通过一个唯一的 thread_id 分组。所有具有相同 thread_id 的 trace 会聚合在一起，在 Opik UI 中显示为一个对话线程。这样，你就能像翻看聊天记录一样，查看用户和 AI 之间的完整对话。
Threads 的典型用途包括：
- 多轮对话：追踪用户和 AI 助手之间的完整聊天会话。
- 用户会话：把单个用户会话中的所有交互归为一组。
- 对话代理：跟踪代理交互和工具使用的流程。
- 工作流追踪：监控跨越多个函数调用的复杂工作流。
thread_id 是用户定义的标识符，必须在项目内唯一。你可以自己生成，也可以用系统提供的 ID。只要保证同一段对话的所有 trace 都使用同一个 thread_id，Opik 就能自动把它们串起来。
如何记录对话？
记录对话的核心就是在创建 trace 时指定 thread_id。Opik 支持多种方式，包括低级 SDK、Python 装饰器和各种集成库。下面我们逐一看看。
TypeScript SDK
如果你用 TypeScript，可以在创建 trace 时直接传入 threadId：
import { Opik } from "opik";
const client = new Opik({
apiUrl: "https://www.comet.com/opik/api",
apiKey: "your-api-key",
projectName: "your-project-name",
workspaceName: "your-workspace-name",
});
const threadId = "your-thread-id";
const trace = client.trace({
name: "chat turn",
input: { user: "Hi there" },
output: { assistant: "Hello!" },
threadId
});
这样，这次 trace 就被归到了指定的线程下。
Python 装饰器
Python 用户可以用 @opik.track 装饰器，并通过 opik_args 参数设置 thread_id：
import opik
from opik import opik_context
@opik.track
def chat_message(input):
return "Opik is an Open Source GenAI platform"
chat_message("What is Opik ?", opik_args={"trace": {"thread_id": "f174a"}})
也可以使用 opik_context 模块动态更新：
@opik.track
def chat_message(input):
thread_id = "f174a"
opik_context.update_current_trace(thread_id=thread_id)
return "Opik is an Open Source GenAI platform"
chat_message("What is Opik ?", thread_id)
Python SDK
如果直接用 Python SDK，在 client.trace() 里传入 thread_id 即可：
import opik
opik_client = opik.Opik()
thread_id = "55d84"
trace = opik_client.trace(
name="chat_conversation",
input="What is Opik?",
output="Opik is an Open Source GenAI platform",
thread_id=thread_id
)
LangGraph 集成
LangGraph 会自动把它自己的 thread_id 映射为 Opik 的 thread_id，所以配置起来非常自然：
from langgraph.graph import StateGraph
from opik.integrations.langchain import OpikTracer
graph = StateGraph(...)
compiled_graph = graph.compile()
thread_id = "user-conversation-789"
config = {
"callbacks": [OpikTracer(project_name="langgraph-conversations")],
"configurable": {"thread_id": thread_id}
}
result1 = compiled_graph.invoke(
{"messages": [{"role": "user", "content": "What is machine learning?"}]},
config=config
)
result2 = compiled_graph.invoke(
{"messages": [{"role": "user", "content": "Can you give me an example?"}]},
config=config
)
两轮对话共享同一个 thread_id，因此会被归入同一个线程。
ADK 集成
ADK 会自动把 session_id 映射为 thread_id：
from opik.integrations.adk import OpikTracer
from google.adk import sessions as adk_sessions, runners as adk_runners
session_service = adk_sessions.InMemorySessionService()
session = session_service.create_session_sync(
app_name="my_chatbot",
user_id="user_123",
session_id="conversation_456"
)
opik_tracer = OpikTracer(project_name="adk-conversations")
runner = adk_runners.Runner(
agent=your_agent,
app_name="my_chatbot",
session_service=session_service
)
result1 = runner.run(
user_id="user_123",
session_id="conversation_456",
new_message="What is machine learning?"
)
result2 = runner.run(
user_id="user_123",
session_id="conversation_456",
new_message="Can you give me an example?"
)
OpenAI Agents
对于 OpenAI Agents，可以使用 trace 上下文管理器的 group_id 来实现线程：
import uuid
from opik import trace
thread_id = str(uuid.uuid4())
with trace(workflow_name="Agent Conversation", group_id=thread_id):
result1 = await Runner.run(agent, "What is machine learning?")
print(result1.final_output)
with trace(workflow_name="Agent Conversation", group_id=thread_id):
new_input = result1.to_input_list() + [
{"role": "user", "content": "Can you give me an example?"}
]
result2 = await Runner.run(agent, new_input)
print(result2.final_output)
注意，每个 trace 的输入会显示为用户消息，输出会显示为 AI 助手响应。这样在 Opik UI 里查看时，就像在看真实的聊天记录。
Thread ID 最佳实践
thread_id 的生成策略取决于你的应用场景。下面几种常见方式供参考。
用户会话基础：把用户 ID 和会话开始时间组合起来，确保每个用户会话有唯一的线程。
import uuid
import opik
user_id = "user_12345"
session_start_time = "2024-01-15T10:30:00Z"
thread_id = f"{user_id}–{session_start_time}"
@opik.track
def process_user_message(message, user_id):
return "Response to: " + message
process_user_message("What is Opik ?", opik_args={"trace": {"thread_id": thread_id}})
UUID 基础：为每段对话生成一个随机 UUID。
import uuid
import opik
thread_id = str(uuid.uuid4())
@opik.track
def start_conversation(initial_message):
return f"Processing: {initial_message}"
start_conversation("What is Opik ?", opik_args={"trace": {"thread_id": thread_id}})
时间戳基础：用时间戳做分组，适合按时间窗口归类的场景。
import time
import opik
thread_id = f"conversation-{int(time.time())}"
@opik.track
def handle_conversation_turn(message):
return f"Response to: {message}"
handle_conversation_turn("What is Opik ?", opik_args={"trace": {"thread_id": thread_id}})
不同的集成对线程的处理方式略有不同。例如，LangChain 可以在 tracer 级别设置 thread_id，也可以通过 metadata 动态传递；LangGraph 自动使用自己的 thread_id；OpenAI Agents 用 group_id；GenAI 和 OpenAI 则通过 opik_args 传入。具体用法可以参考官方文档。
审查对话
对话记录可以在项目级别的 threads 标签页中查看。所有对话都会被追踪，点击 thread ID 就能看到完整对话。线程视图支持 Markdown，方便你阅读返回给用户的内容。如果想深入了解 AI 助手响应是如何生成的，可以点击 View trace 按钮，钻取到具体的 trace 和 span。
在对话视图中，你可以直接给 AI 助手的回复点赞或点踩。这个反馈分数会被记录，并关联到对应的 trace。切换到 trace 视图后，你还可以查看完整 trace，并通过注释功能添加更多反馈分数。
对话评分
你可以随时给线程分配对话级别的反馈分数。线程是由追踪代理或通过 thread_id 相互连接的 trace 聚合而成的。在对话列表中，你能看到每个线程关联的反馈分数。
除了评分，你还可以给线程打标签、添加评论。这对于在审查过程中添加上下文或调查特定对话非常有用。
线程在线评分规则冷却期
对于线程级别的在线评估规则（自动评分），Opik 会在线程最后一次活动之后等待一个“冷却期”，然后再运行规则。这给对话留出沉淀的时间，避免在对话还在进行时就匆忙评分。
默认情况下，冷却期是 15 分钟。如果你是自托管版本，可以通过设置 OPIK_TRACE_THREAD_TIMEOUT_TO_MARK_AS_INACTIVE 环境变量来修改。在云版本中，可以在工作区级别修改“Thread online scoring rule cooldown period”。
当新的 trace 被添加到已有线程时，会发生以下事情：
- 现有反馈分数保留：你手动添加的或在线评估生成的分数都会保留。
- 冷却计时器重启：计时器从新 trace 添加的那一刻重新开始，确保在线评估等待完整的冷却期后再对更新后的线程评分。
- 在线评估重新运行：冷却期结束后，线程级别的在线评分规则会自动重新评估完整对话。如果新分数与已有分数同名，旧分数会被更新。
高级线程功能
过滤和搜索线程
你可以在各种 Opik 功能中使用 thread_id 字段过滤线程。在数据导出时，支持的操作符包括 =、!=、contains、not_contains、starts_with、ends_with 以及字典序比较 >、<。在线程评估中，你也可以评估整个对话线程，这对对话质量评估、多轮连贯性评估和用户满意度评分特别有用。
线程管理
线程可以随时添加新的 trace，你也可以在线程上添加反馈分数、评论和标签，无论是否还有新 trace 加入。
程序化线程管理
你还可以用 Opik SDK 以编程方式管理线程。例如，在 Python 中：
import opik
client = opik.Opik()
threads = client.search_traces(
project_name="my-chatbot",
filter_string='thread_id contains "user-session"'
)
for trace in threads:
if trace.thread_id:
thread_content = client.get_trace_content(trace.id)
print(f"Thread: {trace.thread_id}")
print(f"Input: {thread_content.input}")
print(f"Output: {thread_content.output}")
for trace in threads:
trace.log_feedback_score(
name="conversation_quality",
value=0.8,
reason="Good multi-turn conversation flow"
)
这样你就能批量搜索线程、获取内容、添加反馈分数。
下一步
给多轮对话加上可观测性之后，你可以进一步：
Opik 的 Threads 功能让多轮对话的追踪、审查和评分变得简单自然。无论你用什么框架，都能找到合适的集成方式。花点时间把 thread_id 用起来，你会发现对话质量的管理不再是一件头疼的事。
评论前必须登录！
注册