Assistants API

OPENAI的Assistants API允许我们在自己的应用程序中构建AI助手。

助手包含指令(instructions)，并可以利用模型(models)、工具(tools)和文件(files)来响应用户查询。

• Assistants API 当前支持三种类型的工具：代码解释器、文件搜索和函数调用。

工作原理

Assistants API旨在帮助开发者构建功能强大的AI助手，支持执行多种任务。

1. 助手可以向 OpenAI 的模型发送特定指令，从而调整其个性和能力。
2. 助手可以同时访问多个工具，这些工具既可以是 OpenAI 托管的工具（如code_interpreter和file_search）。
   1. 也可以是我们自行构建或托管的工具（通过函数调用function calling实现）。
3. 助手支持访问具有持久性的thread。
   1. thread通过存储消息历史记录并在对话超出模型上下文长度时进行截断，从而简化了AI应用程序的开发。
   2. 只需创建一次thread，并在用户回复时不断向其中追加消息即可。
4. 助手可以访问多种格式的文件——这些文件既可以在助手创建时提供，也可以作为助手与用户之间thread的一部分提供。
   1. 在使用工具时，助手还可以创建文件（例如图像、电子表格等），并在其生成的消息中引用所使用的文件。

快速入门

Assistants API的典型集成流程如下：

1. 通过定义自定义指令并选择模型来创建Assistant。
   1. 如果需要，可以添加文件并启用诸如代码解释器、文件搜索和函数调用等工具。
2. 当用户开始对话时，创建一个Thread。
3. 用户提问时，将Message添加到该Thread。
4. 在该Thread上运行Assistant，通过调用模型和工具生成响应。

创建一个Assistant：

Assistant代表一个实体，可以通过配置多个参数（如模型mode、指令instructions和工具tools）来响应用户的消息。

from openai import OpenAI
client = OpenAI()

assistant = client.beta.assistants.create(
  name="Math Tutor",
  instructions="You are a personal math tutor. Write and run code to answer math questions.",
  tools=[{"type": "code_interpreter"}],
  model="gpt-4o",
)

创建一个Thread：

Thread代表用户与一个或多个助理之间的交流。

当用户（或你的AI应用）开始与Assistant交流时，可以创建一个Thread。

thread = client.beta.threads.create()

添加Message到Thread中：

用户或应用程序创建的消息会作为Message对象添加Thread中。

• 消息可以包含文本和文件。

每个Thread最多可以容纳100,000 条消息，对于超出模型上下文窗口限制的部分，会自动智能截取超出的消息内容。

message = client.beta.threads.messages.create(
  thread_id=thread.id,
  role="user",
  content="I need to solve the equation `3x + 11 = 14`. Can you help me?"
)

创建Run：

可以使用Python和Node SDK中的创建和流式传输助手来创建Run并流式传输响应。

from typing_extensions import override
from openai import AssistantEventHandler
 
# First, we create a EventHandler class to define
# how we want to handle the events in the response stream.
 
class EventHandler(AssistantEventHandler):    
  @override
  def on_text_created(self, text) -> None:
    print(f"\nassistant > ", end="", flush=True)
      
  @override
  def on_text_delta(self, delta, snapshot):
    print(delta.value, end="", flush=True)
      
  def on_tool_call_created(self, tool_call):
    print(f"\nassistant > {tool_call.type}\n", flush=True)
  
  def on_tool_call_delta(self, delta, snapshot):
    if delta.type == 'code_interpreter':
      if delta.code_interpreter.input:
        print(delta.code_interpreter.input, end="", flush=True)
      if delta.code_interpreter.outputs:
        print(f"\n\noutput >", flush=True)
        for output in delta.code_interpreter.outputs:
          if output.type == "logs":
            print(f"\n{output.logs}", flush=True)
 
# Then, we use the `stream` SDK helper 
# with the `EventHandler` class to create the Run 
# and stream the response.
 
with client.beta.threads.runs.stream(
  thread_id=thread.id,
  assistant_id=assistant.id,
  instructions="Please address the user as Jane Doe. The user has a premium account.",
  event_handler=EventHandler(),
) as stream:
  stream.until_done()

Run是异步的，这意味着需要通过轮询Run对象来监控其状态status，直到达到终止状态。

为了方便，创建和轮询（create and poll）SDK助手可以帮助您创建Run并轮询其完成状态。

run = client.beta.threads.runs.create_and_poll(
  thread_id=thread.id,
  assistant_id=assistant.id,
  instructions="Please address the user as Jane Doe. The user has a premium account."
)

一旦Run完成，就可以列出Assistant添加到Thread中的Messages。

if run.status == 'completed': 
  messages = client.beta.threads.messages.list(
    thread_id=thread.id
  )
  print(messages)
else:
  print(run.status)

如果想查看在此运行期间所做的任何工具调用，还可以列出此运行（Run）的步骤（Run Steps）。