Cookbook: 使用 Litefuse 为 OpenAI Assistants API 提供可观测性
本 cookbook 演示如何使用 Litefuse 的 observe 装饰器 对发送给 OpenAI Assistants API 的调用进行 trace。内容涵盖：创建 assistant、在 thread 上运行它，并通过 Litefuse trace 观察执行过程。
注意：原生的 OpenAI SDK wrapper 不支持对 OpenAI assistants API 的 trace，你需要按照本 notebook 的方式通过装饰器进行 instrument。
什么是 Assistants API？
OpenAI 的 Assistants API 让开发者可以构建能并行使用多种工具和数据源的 AI assistant，例如代码解释器、文件检索以及通过函数调用创建的自定义工具。这些 assistant 可以使用特定 prompt 访问 OpenAI 的语言模型（如 GPT-4），维护持续的对话历史，并处理文本、图像、电子表格等多种文件格式。开发者可以基于自有数据对语言模型进行微调，并控制输出随机性等参数。该 API 提供了一个将语言理解与外部工具及数据相结合的 AI 应用框架。
示例 Trace 输出
设置
安装所需的依赖：
注意： 本指南使用我们的 Python SDK v2。我们已经推出了基于 OpenTelemetry 的全新改进版 SDK，请查看 SDK v3，体验更强大、更易用的 SDK。
%pip install --upgrade openai "langfuse<3.0.0"设置环境变量：
import os
# 从项目设置页面获取你的项目密钥
# https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"] = "https://litefuse.cloud"
# 你的 openai key
os.environ["OPENAI_API_KEY"] = ""分步说明
1. 创建 Assistant
使用 client.beta.assistants.create 方法创建一个新的 assistant。你也可以通过 OpenAI 控制台来创建：
from langfuse import observe
from openai import OpenAI
@observe()
def create_assistant():
client = OpenAI()
assistant = client.beta.assistants.create(
name="Math Tutor",
instructions="You are a personal math tutor. Answer questions briefly, in a sentence or less.",
model="gpt-4"
)
return assistant
assistant = create_assistant()
print(f"Created assistant: {assistant.id}")2. 运行 Assistant
创建一个 thread 并在其上运行 assistant：
@observe()
def run_assistant(assistant_id, user_input):
client = OpenAI()
thread = client.beta.threads.create()
client.beta.threads.messages.create(
thread_id=thread.id, role="assistant", content="I am a math tutor that likes to help math students, how can I help?"
)
client.beta.threads.messages.create(
thread_id=thread.id, role="user", content=user_input
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant_id,
)
return run, thread
user_input = "I need to solve the equation `3x + 11 = 14`. Can you help me?"
run, thread = run_assistant(assistant.id, user_input)
print(f"Created run: {run.id}")3. 获取响应
从 thread 中获取 assistant 的响应：
import json
from langfuse import get_client
langfuse = get_client()
@observe()
def get_response(thread_id, run_id):
client = OpenAI()
messages = client.beta.threads.messages.list(thread_id=thread_id, order="asc")
assistant_response = messages.data[-1].content[0].text.value
# get run for token counts
run_log = client.beta.threads.runs.retrieve(
thread_id=thread_id,
run_id=run_id
)
message_log = client.beta.threads.messages.list(
thread_id=thread_id,
)
input_messages = [{"role": message.role, "content": message.content[0].text.value} for message in message_log.data[::-1][:-1]]
# log internal generation within the openai assistant as a separate child generation to Litefuse
# get langfuse client used by the decorator, uses the low-level Python SDK
langfuse_client = langfuse.client_instance
# pass trace_id and current observation ids to the newly created child generation
langfuse_client.generation(
trace_id=langfuse.get_current_trace_id(),
parent_observation_id=langfuse.get_current_observation_id(),
model=run.model,
usage_details=run.usage,
input=input_messages,
output=assistant_response
)
return assistant_response, run
response = get_response(thread.id, run.id)
print(f"Assistant response: {response[0]}")完整流程合并到一个 trace 中
import time
@observe()
def run_math_tutor(user_input):
assistant = create_assistant()
run, thread = run_assistant(assistant.id, user_input)
time.sleep(5) # notebook only, wait for the assistant to finish
response = get_response(thread.id, run.id)
return response[0]
user_input = "I need to solve the equation `3x + 11 = 14`. Can you help me?"
response = run_math_tutor(user_input)
print(f"Assistant response: {response}")这个 Litefuse trace 展示了创建 assistant、在 thread 上以用户输入运行它、获取响应的完整流程，并包含捕获到的输入/输出数据。
了解更多
如果你使用非 Assistants API 的接口，可以通过 OpenAI SDK wrapper 进行 trace。更多详情请查看 Litefuse 文档。