十四、OpenAI之助手API(Asistants API)

助手API允许你在自己的应用系统中构建一个AI助手。助手有指令，能利用模型、工具和文件响应用户的查询。助手API目前支持3种类型的工具：代码交互，文件搜索和函数调用。
你可以使用助手后台探索助手的能力，或通过这个指南的大纲一步步地构建

概述

一个典型的助手API整合包含以下步骤：

1. 通过定义自定义指令并选择模型来创建助手。如果有帮助，添加文件并启用代码解释器、文件搜索和函数调用等工具。
2. 当用户开始对话时创建一个线程。
3. 在用户提问时向线程添加消息。
4. 在线程上运行助手，通过调用模型和工具来生成响应。

本入门指南介绍了创建和运行使用代码解释器的助手的关键步骤。在本例中，我们将创建一个助手，它是一个个人数学导师，启用了代码解释器工具。

步骤1：创建一个助手
Assistant代表一个实体，可以将其配置为使用模型、指令和工具等多个参数响应用户的消息。

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",
)

步骤2：创建一个线程
线程表示用户与一个或多个助手之间的对话。当用户(或你的AI应用程序)开始与你的助手对话时，你可以创建一个线程。

thread = client.beta.threads.create()

步骤3：添加一个消息到线程中
用户或应用程序创建的消息的内容作为消息对象添加到线程中。消息可以同时包含文本和文件。你可以添加到线程的消息数量没有限制——我们智能地截断任何不适合模型上下文窗口的上下文。

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?"
)

步骤4：创建一个运行器
一旦将所有用户消息添加到线程中，您就可以使用任何助手运行线程。创建运行使用与助手关联的模型和工具来生成响应。这些响应作为辅助消息添加到线程中。
你可以使用Python和Node sdk中的’create and stream’帮助来创建一个运行并流式处理响应。

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()