My Little World

AssistantsAPI

发表于

Assistants API

https://platform.openai.com/docs/assistants/overview

Assistants API 的主要能力

已有能力：

  1. 创建和管理 assistant，每个 assistant 有独立的配置
  2. 支持无限长的多轮对话，对话历史保存在 OpenAI 的服务器上
  3. 通过自有向量数据库支持基于文件的 RAG
  4. 支持 Code Interpreter
    a. 在沙箱里编写并运行 Python 代码
    b. 自我修正代码
    c. 可传文件给 Code Interpreter
  5. 支持 Function Calling
  6. 支持在线调试的 Playground

承诺未来会有的能力：

  1. 支持 DALL·E
  2. 支持图片消息
  3. 支持自定义调整 RAG 的配置项

收费：

  1. 按 token 收费。无论多轮对话，还是 RAG，所有都按实际消耗的 token 收费
  2. 如果对话历史过多超过大模型上下文窗口，会自动放弃最老的对话消息
  3. 文件按数据大小和存放时长收费。1 GB 向量存储 一天收费 0.10 美元
  4. Code interpreter 跑一次 $0.03

划重点：使用 assistant 的意义之一，是可以隔离不同角色的 instruction 和 function 能力。
可以为每个应用，甚至应用中的每个有对话历史的使用场景，创建一个 assistant。

1
2
3
4
5
assistant = client.beta.assistants.create(
    name="Demo test",
    instructions="你叫xxxx, 你负责xxxx",
    model="gpt-4o",
)

管理 thread

Threads：

  1. Threads 里保存的是对话历史，即 messages
  2. 一个 assistant 可以有多个 thread
  3. 一个 thread 可以有无限条 message
  4. 一个用户与 assistant 的多轮对话历史可以维护在一个 thread 里
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 可以根据需要，自定义 `metadata`，比如创建 thread 时，把 thread 归属的用户信息存入。也可以不传
thread = client.beta.threads.create(
    metadata={"fullname": "xxx", "username": "hhh"}
)

===>{
    "id": "thread_ZO9PbLBA4sHJ1xrnKwhprusi",
    "created_at": 1718986823,
    "metadata": {
        "fullname": "xxx",
        "username": "hhh"
    },
    "object": "thread",
    "tool_resources": {
        "code_interpreter": {
            "file_ids": []
        },
        "file_search": null
    }
}

Thread ID 如果保存下来，是可以在下次运行时继续对话的。
从 thread ID 获取 thread 对象的代码

1
thread = client.beta.threads.retrieve(thread.id)

此外，还有：

  1. threads.modify() 修改 thread 的 metadata和tool_resources
  2. threads.retrieve() 获取 thread
  3. threads.delete() 删除 thread。

具体文档参考：https://platform.openai.com/docs/api-reference/threads

给 Threads 添加 Messages

这里的 messages 结构要复杂一些：

  1. 不仅有文本，还可以有图片和文件
  2. 也有metadata
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

message = client.beta.threads.messages.create(
    thread_id=thread.id,  # message 必须归属于一个 thread
    role="user",          # 取值是 user 或者 assistant。但 assistant 消息会被自动加入，我们一般不需要自己构造
    content="你都能做什么？",
)

===>{
    "id": "msg_P2lX3QL5arxOO8tg4JoAIrhb",
    "assistant_id": null,
    "attachments": [],
    "completed_at": null,
    "content": [
        {
            "text": {
                "annotations": [],
                "value": "你都能做什么？"
            },
            "type": "text"
        }
    ],
    "created_at": 1718887069,
    "incomplete_at": null,
    "incomplete_details": null,
    "metadata": {},
    "object": "thread.message",
    "role": "user",
    "run_id": null,
    "status": null,
    "thread_id": "thread_ZO9PbLBA4sHJ1xrnKwhprusi"
}

还有如下函数：

  1. threads.messages.retrieve() 获取 message
  2. threads.messages.update() 更新 message 的 metadata
  3. threads.messages.list() 列出给定 thread 下的所有 messages

具体文档参考：https://platform.openai.com/docs/api-reference/messages

也可以在创建 thread 同时初始化一个 message 列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
thread = client.beta.threads.create(
    messages=[
        {
            "role": "user",
            "content": "你好",
        },
        {
            "role": "assistant",
            "content": "有什么可以帮您？",
        },
        {
            "role": "user",
            "content": "你是谁？",
        },
    ]
)

开始 Run

  1. 用 run 把 assistant 和 thread 关联，进行对话
  2. 一个 prompt 就是一次 run

（执行一次run, 如果run 入参的thread 里面携带(有绑定)message 就相当于向LLM 进行一次提问）

1
2
3
4
5
6
7
8
9
10
11
12
13
assistant_id = "asst_ahXpE6toS71zFyq9h4iMIDj2"  # 从 Playground 中拷贝

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

Run 的底层是个异步调用，意味着它不等大模型处理完，就返回。我们通过 run.status了解大模型的工作进展情况，来判断下一步该干什么。

run.status 有的状态，和状态之间的转移关系如图。

流式运行

  1. 创建回调函数
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from typing_extensions import override
from openai import AssistantEventHandler


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)

更多流中的 Event： https://platform.openai.com/docs/api-reference/assistants-streaming/events

  1. 运行 run
1
2
3
4
5
6
7
8
9
10
11
12
13
# 添加新一轮的 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="你说什么？",
)
# 使用 stream 接口并传入 EventHandler
with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant_id,
    event_handler=EventHandler(),
) as stream:
    stream.until_done()

还有如下函数：

  1. threads.runs.list() 列出 thread 归属的 run
  2. threads.runs.retrieve() 获取 run
  3. threads.runs.update() 修改 run 的 metadata
  4. threads.runs.cancel() 取消 in_progress 状态的 run

具体文档参考：https://platform.openai.com/docs/api-reference/runs

使用 Tools

创建 Assistant 时声明 Code_Interpreter

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
如果用代码创建：
assistant = client.beta.assistants.create(
    name="Demo Assistant",
    instructions="你是人工智能助手。你可以通过代码回答很多数学问题。",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4o"
)
在回调中加入 code_interpreter 的事件响应
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)

    @override
    def on_tool_call_created(self, tool_call):
        """响应工具调用"""
        print(f"\nassistant > {tool_call.type}\n", flush=True)

    @override
    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)

发个 Code Interpreter 请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 创建 thread
thread = client.beta.threads.create()

# 添加新一轮的 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="用代码计算 1234567 的平方根",
)
# 使用 stream 接口并传入 EventHandler
with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant_id,
    event_handler=EventHandler(),
) as stream:
    stream.until_done()

Code_Interpreter 操作文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 上传文件到 OpenAI
file = client.files.create(
    file=open("mydata.csv", "rb"),
    purpose='assistants'
)

# 创建 assistant
my_assistant = client.beta.assistants.create(
    name="CodeInterpreterWithFileDemo",
    instructions="你是数据分析师，按要求分析数据。",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}],
    tool_resources={
        "code_interpreter": {
          "file_ids": [file.id]  # 为 code_interpreter 关联文件
        }
    }
)

关于文件操作，还有如下函数：

  1. client.files.list() 列出所有文件
  2. client.files.retrieve() 获取文件对象
  3. client.files.delete() 删除文件
  4. client.files.content() 读取文件内容

具体文档参考：https://platform.openai.com/docs/api-reference/files

创建 Assistant 时声明 Function calling

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
assistant = client.beta.assistants.create(
  instructions="你叫瓜瓜。你是AGI课堂的助手。你只回答跟AI大模型有关的问题。不要跟学生闲聊。每次回答问题前，你要拆解问题并输出一步一步的思考过程。",
  model="gpt-4o",
  tools=[{
    "type": "function",
    "function": {
      "name": "course_info",
      "description": "用于查看具体课程信息，包括时间表，题目，讲师，等等。Function输入必须是一个合法的SQL表达式。",
      "parameters": {
        "type": "object",
        "properties": {
          "query": {
            "type": "string",
            "description": "SQL query extracting info to answer the user's question.\nSQL should be written using this database schema:\n\nCREATE TABLE Courses (\n\tid INT AUTO_INCREMENT PRIMARY KEY,\n\tcourse_date DATE NOT NULL,\n\tstart_time TIME NOT NULL,\n\tend_time TIME NOT NULL,\n\tcourse_name VARCHAR(255) NOT NULL,\n\tinstructor VARCHAR(255) NOT NULL\n);\n\nThe query should be returned in plain text, not in JSON.\nThe query should only contain grammars supported by SQLite."
          }
        },
        "required": [
          "query"
        ]
      }
    }
  }]
)

两个无依赖的 function 会在一次请求中一起被调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 创建 thread
thread = client.beta.threads.create()

# 添加 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Q1，Q2", ===> 两个问题对应两个函数，一起被调用
)
# 使用 stream 接口并传入 EventHandler
with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant.id,
    event_handler=EventHandler(),
) as stream:
    stream.until_done()

tool.type 为file_search时相当于内置的 RAG 功能

创建 Vector Store，上传文件

  1. 通过代码创建 Vector Store
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
vector_store = client.beta.vector_stores.create(
  name="MyVectorStore"
)

通过代码上传文件到 OpenAI 的存储空间
file = client.files.create(
  file=open("agiclass_intro.pdf", "rb"),
  purpose="assistants"
)

通过代码将文件添加到 Vector Store
vector_store_file = client.beta.vector_stores.files.create(
  vector_store_id=vector_store.id,
  file_id=file.id
)

批量上传文件到 Vector Store
files = ['file1.pdf','file2.pdf']

file_batch = client.beta.vector_stores.file_batches.upload_and_poll(
    vector_store_id=vector_store.id,
    files=[open(filename, "rb") for filename in files]
)

Vector store 和 vector store file 也有对应的 list,retrieve 和 delete等操作。

具体文档参考：

  1. Vector store: https://platform.openai.com/docs/api-reference/vector-stores
  2. Vector store file: https://platform.openai.com/docs/api-reference/vector-stores-files
  3. Vector store file 批量操作: https://platform.openai.com/docs/api-reference/vector-stores-file-batches

创建 Assistant 时声明 RAG 能力

RAG 实际被当作一种 tool

1
2
3
4
5
assistant = client.beta.assistants.create(
  instructions="你是个问答机器人，你根据给定的知识回答用户问题。",
  model="gpt-4o",
  tools=[{"type": "file_search"}],
)

指定检索源

1
2
3
4
assistant = client.beta.assistants.update(
  assistant_id=assistant.id,
  tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)

RAG 请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 创建 thread
thread = client.beta.threads.create()

# 添加 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="AI⼤模型全栈⼯程师适合哪些人",
)
# 使用 stream 接口并传入 EventHandler
with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant_id,
    event_handler=EventHandler(),
) as stream:
    stream.until_done()

多个 Assistants 协作

划重点： 使用 assistant 的意义之一，是可以隔离不同角色的 instruction 和 function 能力。

6顶思维帽实验

技术选型参考

GPTs 现状：

  1. 界面不可定制，不能集成进自己的产品
  2. 只有 ChatGPT Plus/Team/Enterprise 用户才能访问
  3. 未来开发者可以根据使用量获得报酬，北美先开始
  4. 承诺会推出 Team/Enterprise 版的组织内部专属 GPTs

适合使用 Assistants API 的场景：

  1. 定制界面，或和自己的产品集成
  2. 需要传大量文件
  3. 服务国外用户，或国内 B 端客户
  4. 数据保密性要求不高
  5. 不差钱

适合使用原生 API 的场景：

  1. 需要极致调优
  2. 追求性价比
  3. 服务国外用户，或国内 B 端客户
  4. 数据保密性要求不高

适合使用国产或开源大模型的场景：

  1. 服务国内用户
  2. 数据保密性要求高
  3. 压缩长期成本
  4. 需要极致调优