Hacker and Geeker's Way

上次我们介绍了 OpenAI 的新版 API，包括语音转文字、生成图片和图片识别等功能，这次 API 的更新还包含了一个重量级的功能，就是类似 GPTs 的 Assistant API，它不仅可以完成 GPTs 的所有功能，还能使用自定义的工具，可以说是比 GPTs 更加强大。今天我们就来介绍 Assistant API 的基本原理和使用方法，最后通过一些代码示例来展示它的强大功能。

设计原理

上面是整理的 Assistant API 对象关系图，在图中我们可以看到如下关系：

• Assistant 对象是用来执行命令的对象，它有多个属性，其中包括 tools 和 file_ids，分别对应 Tool 对象和 File 对象。
• Thread 对象表示一个聊天会话，它是有状态的，就像 ChatGPT 网页上的每个历史记录，我们可以对历史记录进行重新对话，它包含了多个 Message 对象。
• Message 对象表示一条聊天消息，分不同角色的消息，包括 user、assistant 和 tool 等。
• Run 对象表示一次指令执行的过程，需要指定执行命令的对象 Assistant 和聊天会话 Thread，一个 Thread 可以创建多个 Run。
• Run Step 对象表示执行的步骤，一个 Run 包含多个 Run Step。
• Tool 对象表示执行命令时需要用到的工具，有代码解释器（Code Interpreter）、知识检索（Retrieval）和自定义工具（Function Calling）等。
• File 对象表示执行命令时所用到的文件，比如知识检索时上传的文档，它可以一开始在 Assistant 对象中指定，也可以聊天过程中在 Message 对象中指定。

API 执行过程如下：

• 创建 Assistant
• 创建 Thread 和 Message，可以分开创建也可以一起创建
• 创建 由 Assistant 和 Thread 组成的 Run，创建完 Run 后会自动执行 Thread 中的指令
• 轮询 Run 状态，检查是否为 completed
• （可选）如果是调用自定义工具，需要提交工具的执行结果
• 如果状态是 completed 则获取最终结果

Assistant

首先我们需要创建一个 Assistant 对象，因为这个 API 是 beta 版本，如果是通过 curl 调用 API 的话，需要在 header 中加上OpenAI-Beta: assistants=v1，如果是使用 OpenAI 的 Python 或 Npm 包的话则不需要，这些包已经默认帮你添加了。示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

from openai import OpenAI

client = OpenAI()

def create_assistants(instructions, tools=[], file_ids=[]):
    assistant = client.beta.assistants.create(
        name="Help assistant",
        instructions=instructions,
        tools=tools,
        model="gpt-3.5-turbo-1106",
        file_ids=file_ids,
    )
    return assistant

• 因为是 beta 版本，model 参数需要用最新的模型，比如 gpt-3.5-turbo-1106 或者 gpt-4-1106-preview。
• name 是 Assistant 的名字。
• instructions 参数相当是 Chat API 中的 system message，即系统指令，一般用来让 LLM（大语言模型）扮演某种角色。
• tools 是指使用的工具，有代码解释器、知识检索和自定义工具等。
• file_ids 是指上传的文件，在每个 Assistant 中最多添加 20 个文件，每个文件最大 512M，整个组织的文件量不能超过 100G，文件的支持类型可以参考这里。

上传文件的方法如下：

1
2
3

def create_file(file_path):
    file = client.files.create(file=open(file_path, "rb"), purpose="assistants")
    return file

• 目前上传文件有 2 种用途，一种是用于微调（fine turning），一种是用于 Assistant，因此在 Assistant 中上传的文件，purpose 要写 assistants。

Assistant 和 File 更多的 API 可以看官方的 API 文档。

Thread

接下来是创建一个 Thread，可以单独创建，也可以和 Message 一起创建，这里我们连同 Message 一起创建，示例代码如下：

1
2
3
4
5
6
7
8
9
10
11

def create_thread(prompt):
    thread = client.beta.threads.create(
        messages=[
            {
                "role": "user",
                "content": prompt,
                "file_ids": [file.id],
            }
        ]
    )
    return thread

• 在 Thread 中没有限制 Message 的数量，但一旦超过 token 限制，就会进行智能截断。
• prompt 是指用户输入的问题。
• file_ids 是指上传的文件，可以包含图片和文件，但目前 user 角色的消息还不支持图片。

Thread 更多的 API 可以看官方的 API 文档。

Run

然后是创建 Run，创建 Run 时需要指定 Assistant 和 Thread，示例代码如下：

1
2
3
4
5
6

def run_assistant(thread, assistant):
    run = client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
    )
    return run

• 在创建 Run 时，可以通过 model、instructions、tools 等参数来覆盖 Assistant 中的设置，但 file_ids 不能被覆盖。

下面是 Run 的状态流转图：

• 当创建 Run 后，Run 会自动进入 queued 状态
• queued 状态后进入到 in_process 状态，表示 Run 正在执行
• 根据执行结果，如果执行成功则进入 completed 状态，如果执行失败则进入 failed 状态
• 在 in_process 状态下，可以通过 Run 的 cancel API 来取消执行，Run 会进入 cancelling 状态，然后进入 cancelled 状态
• 在执行过程中如果用到了自定义工具，in_process 状态会进入 requireds_action 状态，表示需要提交工具的执行结果
• 提交工具的执行结果后，Run 会再次进入 queued 状态
• 如果迟迟没有提交工具的执行结果，超过了过期时间（一般是从创建 Run 时算起 10 分钟），Run 会进入 expired 状态
• 当状态为 in_process 时，表示 Thread 被锁定，这意味着 Thread 不能添加新消息和创建新的 Run

当创建完了 Run 后，我们需要根据 Run 的 ID 来获取 Run，查询其状态，这是一个重复的过程，下面是查询 Run 的方法：

1
2
3

def retrieve_run(thread, run):
    run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
    return run

Run 更多的 API 可以看官方的 API 文档。

Run Steps

当 Run 运行完成后，我们可以通过获取 Run 的步骤来查看执行的过程，示例代码如下：

1
2
3
4
5
6

def list_run_steps(thread, run):
    run_steps = client.beta.threads.runs.steps.list(
        thread_id=thread.id,
        run_id=run.id,
    )
    return run_steps

• 获取的 Run Steps 是按时间从早到晚进行排序的，即最早的 Run Step 在最前面，最晚的 Run Step 在最后面。
• Run Step 有两种类型，一种是 message_creation，表示创建消息，另一种是 tool_calls，表示执行工具，包括代码解释器、知识检索和自定义工具等。
• 获取的结果包含分页的参数，可以通过 has_more 来判断是否有下一页的数据。

Run Step 也有对应的状态，状态流转图如下所示：

Run Step 状态比 Run 的状态要简单一些，状态的流转条件跟 Run 的一样，这里就不再赘述了。

Message

当 Run 运行完成后，我们还需要获取 message 的结果，示例代码如下：

1
2
3

def list_messages(thread):
    thread_messages = client.beta.threads.messages.list(thread_id=thread.id)
    return thread_messages

• 获取的 message 是按时间倒序排序的，即最新的 message 在最前面，最早的 message 在最后面。因此我们要获取 Run 的最终答案，只需要获取第一条 message 的结果即可。
• 获取的结果包含分页的参数，可以通过 has_more 来判断是否有下一页的数据。

Message 更多的 API 可以看官方的 API 文档。

代码演示

下面我们来通过几个示例来演示下 Assistant API 的具体功能。

代码解释器

我们使用代码解释器来解一道方程式，示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from time import sleep

def code_interpreter():
    assistant = create_assistants(
        instructions="你是一个数学导师。写代码来回答数学问题。",
        tools=[{"type": "code_interpreter"}]
    )
    thread = create_thread(prompt="我需要计算这个方程式的解：`3x + 11 = 14`。你能帮我吗？")
    run = run_assistant(thread, assistant)
    while True:
        run = retrieve_run(thread=thread, run=run)
        if run.status == "completed":
            break
        sleep(1)
    messages = list_messages(thread)
    print(messages.data[0].content[0].text.value)
    print(f"messages: {messages.json()}")
    run_steps = list_run_steps(thread=thread, run=run)
    print(f"run_steps: {run_steps.json()}")

• 我们首先创建一个 Assistant，指定 instructions 为你是一个数学导师。写代码来回答数学问题。
• Assistant 中使用了代码解释器工具{"type": "code_interpreter"}。
• 然后创建一个 Thread，输入我们的问题，求解一道方程式。
• 创建 Run，然后通过一个循环来查询 Run 的状态，直到状态为 completed 时退出循环。
• 获取最终的结果，即第一条 message 的内容。

运行结果如下：

1

方程式`3x + 11 = 14`的解为 x = 1。

我们再来看这个 Run 中产生的所有 Messages 信息：

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

{
  "data": [
    {
      "id": "msg_7cyjjNTgjXOtWlnLOFIeMKW4",
      "object": "thread.message",
      "role": "assistant",
      "content": [
        {
          "text": {
            "annotations": [],
            "value": "方程式`3x + 11 = 14`的解为x = 1。"
          },
          "type": "text"
        }
      ]
    },
    {
      "id": "msg_xVpdPAd4VOO6Ve5bJ5XoOoiw",
      "object": "thread.message",
      "role": "user",
      "content": [
        {
          "text": {
            "annotations": [],
            "value": "我需要计算这个方程式的解：`3x + 11 = 14`。你能帮我吗？"
          },
          "type": "text"
        }
      ]
    }
  ],
  "object": "list",
  "has_more": false
}

总共只有 2 条消息，一条是 user 输入的问题，另一条是 assistant 返回的结果，中间并没有工具的消息。我们再来看这个 Run 中的所有 Run Step 信息：

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

{
  "data": [
    {
      "object": "thread.run.step",
      "status": "completed",
      "step_details": {
        "message_creation": { "message_id": "msg_7cyjjNTgjXOtWlnLOFIeMKW4" },
        "type": "message_creation"
      },
      "type": "message_creation"
    },
    {
      "object": "thread.run.step",
      "status": "completed",
      "step_details": {
        "tool_calls": [
          {
            "id": "call_mTRfGO52jA6oPLLMACKr5HD5",
            "code_interpreter": {
              "input": "from sympy import symbols, Eq, solve\r\n\r\n# Define the variable\r\nx = symbols('x')\r\n\r\n# Define the equation\r\nequation = Eq(3*x + 11, 14)\r\n\r\n# Solve the equation\r\nsolution = solve(equation, x)\r\nsolution",
              "outputs": [{ "logs": "[1]", "type": "logs" }]
            },
            "type": "code_interpreter"
          }
        ],
        "type": "tool_calls"
      },
      "type": "tool_calls"
    }
  ],
  "object": "list",
  "has_more": false
}

这个 Run 有 2 个步骤，一个是创建消息，另外一个是代码解释器的执行，其中代码解释器中执行过程中产生的 input 信息并不会显示到最终的结果中，只是 LLM 的一个思考过程，类似 LangChain 的 Agent 里面的 debug 信息。

知识检索

下面我们再使用知识检索工具来演示一下功能，知识检索工具需要上传一个文件，文件通过 API 上传后，OpenAI 后端会自动分割文档、embedding、存储向量，并提供根据用户问题检索文档相关内容的功能，这些都是自动完成的，用户只需要上传文档即可。我们就随便找一个 pdf 文件来做演示，下面是腾讯云搜的产品文档：

下面是知识检索的示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

def knownledge_retrieve():
    file = create_file("tengxunyun.pdf")
    assistant = create_assistants(
        instructions="你是一个客户支持机器人，请用你的专业知识回答客户的问题。",
        tools=[{"type": "retrieval"}],
        file_ids=[file.id],
    )
    thread = create_thread(prompt="腾讯云云搜是什么")
    run = run_assistant(thread, assistant)
    while True:
        run = retrieve_run(thread=thread, run=run)
        if run.status == "completed":
            break
        sleep(1)
    messages = list_messages(thread)
    print(messages.data[0].content[0].text.value)
    print(f"messages: {messages.json()}")
    run_steps = list_run_steps(thread=thread, run=run)
    print(f"run_steps: {run_steps.json()}")

• 这里我们更换了 Assistant 的 instructions，让它扮演一个客户支持机器人的角色。
• Assistant 中使用了检索工具{"type": "retrieval"}。
• Assistant 中上传了上面的 pdf 文件。
• 然后创建一个 Thread，输入我们的问题，问一个文档相关内容的问题。
• 其他步骤与代码解释器一致。

运行结果如下：

1
2
3
4
5

腾讯云云搜是腾讯云的一站式搜索托管服务平台，提供数据处理、检索串识别、搜索结果获取与排序，搜索数据运营等一整套搜索相关服务。
该平台继承了腾讯 SOSO 在搜索引擎领域多年的技术财富，在搜索架构、海量数据存储和计算、智能排序、用户意图识别、搜索质量运营等方面有很深的技术沉淀。
腾讯云云搜负责了腾讯主要产品的搜索业务，包括微信朋友圈、手机 QQ、腾讯视频、QQ 音乐、应用宝、腾讯地图、QQ 空间等。
它提供数据处理、用户检索串智能识别、排序可定制、高级功能、运营支持等功能，以帮助开发者优化搜索服务，并提供丰富的运营数据查询功能，
包括检索量、文档新增量、检索耗时、检索失败率、热榜等。开发者可以通过腾讯云云搜让自己的搜索更具个性化，更匹配应用的需求。【1†source】

可以看到答案确实是从文档中查找而来，内容基本一致，在结尾处还有一个引用【1†source】，这个是 Message 的注释内容，关于 Message 的注释功能这里不过多介绍，后面有机会再写文章说明，关于 Message 注释功能可以看这里。

我们再来看下知识检索的 Run Step 信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

{
  "data": [
    {
      "type": "message_creation"
      // ......
    },
    {
      "object": "thread.run.step",
      "status": "completed",
      "step_details": {
        "tool_calls": [
          {
            "id": "call_19YklOZJq1HDP8WfmMydcVeq",
            "retrieval": {},
            "type": "retrieval"
          }
        ],
        "type": "tool_calls"
      },
      "thread_id": "thread_ejr0YGodJ2NmG7dFf1hZMPUL",
      "type": "tool_calls"
    }
  ]
}

在检索工具的步骤中，只是返回了工具的类型，但检索的内容并没有放在步骤中，也就是说检索工具并没有产生内部推理过程的信息。

自定义工具

最后我们再来看下自定义工具的示例，我们沿用上一篇文章用到的查询天气工具get_current_weather，我们先定义工具集 tools：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取某个地方当前的天气情况",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市名，比如： 北京, 上海",
                    },
                },
                "required": ["location"],
            },
        },
    }
]

• 工具集只包含一个工具，定义了工具的方法和参数信息

接着我们使用 Assistant API 来调用自定义工具，示例代码如下：

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

def get_current_weather(location):
    """Get the current weather in a given location"""
    if "北京" in location.lower():
        return json.dumps({"location": location, "temperature": "10°"})
    elif "上海" in location.lower():
        return json.dumps({"location": location, "temperature": "15°"})
    else:
        return json.dumps({"location": location, "temperature": "20°"})

def function_calling():
    assistant = create_assistants(
        instructions="你是一个天气机器人，使用提供的工具来回答问题。",
        tools=tools,
    )
    thread = create_thread(prompt="今天北京、上海和成都的天气怎么样？")
    available_functions = {
        "get_current_weather": get_current_weather,
    }
    run = run_assistant(thread, assistant)
    # 下面是处理 Run 并提交工具的返回结果
    # 中间这部分代码在下面演示
    messages = list_messages(thread)
    print(messages.data[0].content[0].text.value)
    print(f"messages: {messages}")
    run_steps = list_run_steps(thread=thread, run=run)
    print(f"run_steps: {run_steps}")

• 和之前 2 个示例不同的地方是，tools 参数用的是我们定义的一个工具集 tools
• 定义了一个字典available_functions 来做函数的动态调用
• 为了演示方便，get_current_weather方法使用一些简单的逻辑来模拟天气查询的功能：

下面是处理 Run 状态并提交工具返回结果的方法，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

while True:
    run = retrieve_run(thread=thread, run=run)
    if run.status == "completed":
        break
    if run.status == "requires_action":
        tool_calls = run.required_action.submit_tool_outputs.tool_calls
        tool_infos = []
        for tool_call in tool_calls:
            function_name = tool_call.function.name
            function_to_call = available_functions[function_name]
            function_args = json.loads(tool_call.function.arguments)
            function_response = function_to_call(
                location=function_args.get("location"),
            )
            tool_infos.append(
                {"call_id": tool_call.id, "function_response": function_response}
            )
        submit_tool_outputs(thread=thread, run=run, tool_infos=tool_infos)
    sleep(1)

• 之前在介绍 Run 状态时说过，当 Assistant 中有自定义工具时，状态从 in_process 会进入 requires_action 状态，表示需要提交工具的执行结果
• 在循环中判断 Run 状态是否为 requires_action，如果是则从 Run 中获取需要执行的工具信息 tool_calls，包括工具名称和参数
• 在示例中会调用 3 次get_current_weather方法，因此我们新建了一个数组 tool_infos 来保存工具返回的结果
• 执行完工具后，将工具返回结果保存到 tool_infos 数组中，同时将 tool_calls 中工具 id 也保存到里面，下面在提交工具结果时会用到

拿到工具返回的结果后，我们通过 Run API 来提交这些结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

def submit_tool_outputs(thread, run, tool_infos):
    tool_outputs = []
    for tool_info in tool_infos:
        tool_outputs.append(
            {
                "tool_call_id": tool_info["call_id"],
                "output": tool_info["function_response"],
            }
        )
    client.beta.threads.runs.submit_tool_outputs(
        thread_id=thread.id,
        run_id=run.id,
        tool_outputs=tool_outputs,
    )

• 这里用到了 Run 提交工具结果的 API，其中 tool_outputs 参数是一个数组，数组中每个元素包含 tool_call_id 和 output 两个属性
• tool_call_id 的值是 tool_calls 中的 id，output 的值是工具的返回结果
• 调用了多少个工具就要在 tool_outputs 添加多少个元素

运行结果如下：

1

今天北京的温度是 10℃，上海的温度是 15℃，成都的温度是 20℃。

改进计划

在 Assistant API 的使用过程中，我们发现现在获取 Run 的状态都是通过轮询的方式，这可能会导致更多的 token 损耗和性能问题，OpenAI 官方介绍在未来会对这一机制进行改进，将其替换成通知模式，另外还有其他改进计划如下：

• 支持流式输出模式（类似 Websocket 或者 SSE）
• 支持通过共享对象状态更新通知来代替轮询
• 支持 DALL-E 3 作为内置的工具
• 支持用户上传图片文件

总结

以上就是 Assistant API 的基本原理和功能介绍，和 GPTs 相比两者各有优势，GPTs 更适合没有编程经验的用户使用，而 Assistant API 则适合有开发经验的开发人员或团队使用，而且可以使用自定义工具来打造更为强大的功能，但也有一些缺点，就是无法使用 DALL-E 3 画图工具和上传图片（这也意味着无法做图片识别），这些功能相信在未来会逐步支持。如果文中有错误或者不足之处，还请在评论区留言。

关注我，一起学习各种人工智能和 AIGC 新技术，欢迎交流，如果你有什么想问想说的，欢迎在评论区留言。