OpenAI之AssistantsAPI

Posted on Edited on In , Views: Word count in article: 7.6k Reading time ≈ 7 mins.

1、快速入门Assistants API

  • Assistants API 允许您在自己的应用程序中构建人工智能助手。一个助手有其指令，并可以利用模型、工具和知识来回应用户查询。

  • Assistants API 目前支持三种类型的工具：

    • 代码解释器 Code Interpreter
    • 检索 Retrieval
    • 函数调用 Function calling

  • 使用Playground可以在线探索和测试 Assistants API 功能。

  • 本入门指南将指导您完成创建和运行使用代码解释器的助手的关键步骤，以下是使用 Assistants API 标准流程：

    • 通过定义其自定义指令并选择 LLM 来创建一个助手(Assistant)。如果有需求，可以添加文件并启用诸如代码解释器、检索和函数调用等工具。
    • 当用户开始对话时，创建一个线程(Thread)。
    • 当用户提问时，向线程添加消息(Messages)。
    • 通过调用模型和工具在线程上运行助手以生成响应。
    OBJECT WHAT IT REPRESENTS
    Assistant 专为特定目的构建的人工智能，使用 OpenAI 的模型并调用工具。
    Thread 助手与用户之间的对话会话。线程存储消息，并自动处理截断，以将内容适应模型的上下文。
    Message 由助手或用户创建的消息。消息可以包括文本、图片和其他文件。消息以列表形式存储在线程上。
    Run 在线程上对一个助手的调用。助手利用其配置和线程的消息执行任务，通过调用模型和工具。作为运行的一部分，助手会将消息追加到线程中。
    Run Step 助手在运行中采取的详细步骤列表。助手可以在其运行期间调用工具或创建消息。检查运行步骤可以让您深入了解助手如何得出最终结果。

2、使用Assistants开发数学辅导老师

  • 在这个示例中，我们正在创建一个数学辅导助手，并启用了代码解释器工具。

    • 创建助手：

      1
      ! pip install -U openai
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      import openai  # 导入 openai 库

      # 从环境变量 OPENAI_API_KEY 中获取 API 密钥
      client = openai.OpenAI()

      # 创建一个名为 "Math Tutor" 的助手，它是一个个人数学辅导老师。这个助手能够编写并运行代码来解答数学问题。
      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",  # 使用模型： GPT-4o
      )

    • 创建线程：一个线程代表用户和一个或多个助手之间的对话。

      1
      2
      # 创建一个交流线程
      thread = client.beta.threads.create()

    • 往线程添加消息：

      • 用户或APP创建的消息内容将作为消息对象（Message Object）添加到线程中。
      • 消息可以包含文本和文件，向线程添加的消息数量没有限制 - OpenAI 会智能地截断任何不适合模型上下文窗口的内容。
      1
      2
      3
      4
      5
      6
      # 在该线程中创建一条用户消息，并提交一个数学问题：“我需要解方程 `3x + 11 = 14`。你能帮忙吗？”
      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?",
      )

    • 调用助手：

      • 一旦所有用户消息都添加到了线程中，你可以使用任何助手运行该线程。
      • 创建一个运行会使用与助手相关的模型和工具来生成响应。这些响应将作为助手消息添加到线程中。
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      # 创建并等待执行流完成，用于处理该线程中的交互和问题解答
      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.",  # 以 Jane Doe 称呼用户，并且用户拥有高级账户
      )

      print("Run completed with status: " + run.status)  # 打印执行流的完成状态

      # 如果执行流状态为 "completed"（已完成），则获取并打印所有消息
      if run.status == "completed":
          messages = client.beta.threads.messages.list(thread_id=thread.id)

          print("\nMessages:\n")
          for message in messages:
              assert message.content[0].type == "text"
              print(f"Role: {message.role.capitalize()}")  # 角色名称首字母大写
              print("Message:")
              print(message.content[0].text.value + "\n")  # 每条消息后添加空行以增加可读性

    • 通过Assistant ID删除指定助手：

      1
      2
      # 删除创建的助手
      client.beta.assistants.delete(assistant.id)

3、使用流式输出实现数学辅导老师

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# 官方案例
from typing_extensions import override
from openai import AssistantEventHandler
import openai

# 从环境变量 OPENAI_API_KEY 中获取 API 密钥
client = openai.OpenAI()

# 创建一个名为 "Math Tutor" 的助手，它是一个个人数学辅导老师。这个助手能够编写并运行代码来解答数学问题。
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-4-1106-preview",  # 使用的模型是 GPT-4
)

# 创建一个交流线程
thread = client.beta.threads.create()

# 在该线程中创建一条消息，表示用户角色，并提交一个数学问题：“我需要解方程 `3x + 11 = 14`。你能帮忙吗？”
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?",
)

print("starting run stream")  # 打印开始执行流的消息

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


# 删除创建的助手
client.beta.assistants.delete(assistant.id)

4、开发Python代码小助手

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import openai  # 导入 openai 库
import time

# 从环境变量 OPENAI_API_KEY 中获取 API 密钥
client = openai.OpenAI()

# 创建一个名为 "Python Master" 的助手，它能根据需求生成可以运行的 Python 代码
assistant_python = client.beta.assistants.create(
    name="Python Master",
    instructions="You are a Python Expert. Generate runnable Python code according to messages.",
    tools=[{"type": "code_interpreter"}],  # 使用工具：代码解释器
    model="gpt-4o",  # 使用模型： GPT-4
)
assistant_python.id # 'asst_oB01W2lzBFO4YQgIn5QldOLd'

# 创建一个交流线程
thread_python = client.beta.threads.create()
thread_python.id # 'thread_glOKBynaIzmlEL2BMDFYIzkt'

# 在该线程中创建一条信息
message = client.beta.threads.messages.create(
    thread_id=thread_python.id,
    role="user",
    content="快速排序咋个写？",
)
message.id # 'msg_FkPrssBSOS5uDEQMGF6IeiHK'

# 创建并等待执行流完成，用于处理该线程中的交互和问题解答
# 方式一：create_and_poll创建并轮询方式
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread_python.id,
    assistant_id=assistant_python.id,
)
print("Run completed with status: " + run.status)  # 打印执行流的完成状态，Run completed with status: completed
run.id # 'run_v7Q9TffTwXB2y7CCPuX5yPdB'

# 方式二：create 和 retrieve方式，判断run状态为（初始为queued）
run2 = client.beta.threads.runs.create(
    thread_id=thread_python.id,
    assistant_id=assistant_python.id,
)
print("Run初始状态 " + run2.status)  # 打印执行流的完成状态，Run初始状态 queued
run2.id # 'run_1vaQWkvblJuY7h3o9oE6kJ4A'
while run2.status == "queued" or run.status == "in_progress":
    run2 = client.beta.threads.runs.retrieve(
        thread_id=thread_python.id,
        run_id=run2.id
    )
    time.sleep(1)
print("Run2 completed with status: " + run2.status)  # 打印执行流的完成状态，Run2 completed with status: completed
# 如果执行流状态为 "completed"（已完成），则获取并打印所有消息
if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread_python.id)

    print("\nMessages:\n")
    for message in messages:
        assert message.content[0].type == "text"
        print(f"Role: {message.role.capitalize()}")  # 角色名称首字母大写
        print("Message:")
        print(message.content[0].text.value + "\n")  # 每条消息后添加空行以增加可读性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 在该线程中创建一条信息
message = client.beta.threads.messages.create(
    thread_id=thread_python.id,
    role="user",
    content="红黑树呢？",
)

# 创建并等待执行流完成，用于处理该线程中的交互和问题解答
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread_python.id,
    assistant_id=assistant_python.id,
)

print("Run completed with status: " + run.status)  # 打印执行流的完成状态，Run completed with status: completed

# 如果执行流状态为 "completed"（已完成），则获取并打印所有消息
if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread_python.id)

    print("\nMessages:\n")
    for message in messages:
        assert message.content[0].type == "text"
        print(f"Role: {message.role.capitalize()}")  # 角色名称首字母大写
        print("Message:")
        print(message.content[0].text.value + "\n")  # 每条消息后添加空行以增加可读性
        
# 删除创建的助手
client.beta.assistants.delete(assistant_python.id)