5、Assistants API

Assistants API

GPTs 和 Assistants API 本质是降低开发门槛

可操控性和易用性之间的权衡与折中：

1. 更多技术路线选择：原生 API、GPTs 和 Assistants API
2. GPTs 的示范，起到教育客户的作用，有助于打开市场
3. 要更大自由度，需要用 Assistants API 开发
4. 想极致调优，还得原生 API + RAG

Assistants API 的主要能力

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

收费：

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

一、GPT Store：创建自己的 GPT

发布链接：https://chat.openai.com/g/g-iU8hVr4jR-wo-de-demogpt

二、Assistants API

!pip install --upgrade openai

from openai import OpenAI
client = OpenAI()

ids = []
assistants = client.beta.assistants.list()
for assistant in assistants:
    ids.append(assistant.id)
# 清理一下教学环境
for id in ids:
    client.beta.assistants.delete(id)

创建一个 Assistant

可以为每个应用，甚至应用中的每个有对话历史的使用场景，创建一个 assistant。

虽然可以用代码创建，也不复杂，例如：

from openai import OpenAI

# 初始化 OpenAI 服务
client = OpenAI()

# 创建助手
assistant = client.beta.assistants.create(
    name="AGIClass Demo",
    instructions="你叫瓜瓜，你是AGI课堂的智能助理。你负责回答与AGI课堂有关的问题。",
    model="gpt-4o",
)

但是，更佳做法是，到 Playground 在线创建，因为：

1. 更方便调整
2. 更方便测试

from openai import OpenAI

# 初始化 OpenAI 服务
client = OpenAI()

# 创建助手
assistant = client.beta.assistants.create(
    name="AGIClass Demo TempLive",
    instructions="你叫瓜瓜，你是AGI课堂的智能助理。你负责回答与AGI课堂有关的问题。",
    model="gpt-4o",
)

print(assistant.id)

asst_xi4KvqarumvNarFA2jdwmzkb

三、代码访问 Assistant

3.1、管理 thread

Threads：

1. Threads 里保存的是对话历史，即 messages
2. 一个 assistant 可以有多个 thread
3. 一个 thread 可以有无限条 message
4. 一个用户与 assistant 的多轮对话历史可以维护在一个 thread 里

import json


def show_json(obj):
    """把任意对象用排版美观的 JSON 格式打印出来"""
    print(json.dumps(
        json.loads(obj.model_dump_json()),
        indent=4,
        ensure_ascii=False
    ))

from openai import OpenAI
import os

from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())

# 初始化 OpenAI 服务
client = OpenAI()   # openai >= 1.3.0 起，OPENAI_API_KEY 和 OPENAI_BASE_URL 会被默认使用

# 创建 thread
thread = client.beta.threads.create()
show_json(thread)

{
    "id": "thread_5EP077dOgvXyJQkbCnbn249q",
    "created_at": 1727162907,
    "metadata": {},
    "object": "thread",
    "tool_resources": {
        "code_interpreter": null,
        "file_search": null
    }
}

可以根据需要，自定义 metadata，比如创建 thread 时，把 thread 归属的用户信息存入。

thread = client.beta.threads.create(
    metadata={"fullname": "王卓然", "username": "wzr"}
)
show_json(thread)

{
    "id": "thread_RWfx9UJ02xTLIfcLQjaxKGeq",
    "created_at": 1727162914,
    "metadata": {
        "fullname": "王卓然",
        "username": "wzr"
    },
    "object": "thread",
    "tool_resources": {
        "code_interpreter": null,
        "file_search": null
    }
}

Thread ID 如果保存下来，是可以在下次运行时继续对话的。

从 thread ID 获取 thread 对象的代码：

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

{
    "id": "thread_RWfx9UJ02xTLIfcLQjaxKGeq",
    "created_at": 1727162914,
    "metadata": {
        "fullname": "王卓然",
        "username": "wzr"
    },
    "object": "thread",
    "tool_resources": {
        "code_interpreter": {
            "file_ids": []
        },
        "file_search": null
    }
}

此外，还有：

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

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

3.2、给 Threads 添加 Messages

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

1. 不仅有文本，还可以有图片和文件
2. 也有 metadata

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

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

还有如下函数：

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 列表

thread = client.beta.threads.create(
    messages=[
        {
            "role": "user",
            "content": "你好",
        },
        {
            "role": "assistant",
            "content": "有什么可以帮您？",
        },
        {
            "role": "user",
            "content": "你是谁？",
        },
    ]
)

show_json(thread)  # 显示 thread
print("-----")
show_json(client.beta.threads.messages.list(
    thread.id))  # 显示指定 thread 中的 message 列表

{
    "id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh",
    "created_at": 1727162936,
    "metadata": {},
    "object": "thread",
    "tool_resources": {
        "code_interpreter": null,
        "file_search": null
    }
}
-----
{
    "data": [
        {
            "id": "msg_bcCprHQl7OGxgZudyc5F9how",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "你是谁？"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "user",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        },
        {
            "id": "msg_JQjoOfqKDYl4RKWNjeQvaCVu",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "有什么可以帮您？"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "assistant",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        },
        {
            "id": "msg_5GfK3Ys1bvyVfkUOoVMSv1ra",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "你好"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "user",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        }
    ],
    "object": "list",
    "first_id": "msg_bcCprHQl7OGxgZudyc5F9how",
    "last_id": "msg_5GfK3Ys1bvyVfkUOoVMSv1ra",
    "has_more": false
}

3.3、开始 Run

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

3.1、直接运行

assistant_id = "asst_psmawyqIV5HrDiwxYAesO4ia"  # 从 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)

{
    "data": [
        {
            "id": "msg_NpWGvI1fbVtUtrRxtTQqecNE",
            "assistant_id": "asst_psmawyqIV5HrDiwxYAesO4ia",
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "我是瓜瓜，AGI课堂的智能助理。我可以帮助您解答与AGI课堂相关的问题，包括课程安排、内容查询等。如果您有任何问题，请随时告诉我！"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162963,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "assistant",
            "run_id": "run_pKbeI1F2KDaLVCpwRRBR8PXt",
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        },
        {
            "id": "msg_bcCprHQl7OGxgZudyc5F9how",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "你是谁？"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "user",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        },
        {
            "id": "msg_JQjoOfqKDYl4RKWNjeQvaCVu",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "有什么可以帮您？"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "assistant",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        },
        {
            "id": "msg_5GfK3Ys1bvyVfkUOoVMSv1ra",
            "assistant_id": null,
            "attachments": [],
            "completed_at": null,
            "content": [
                {
                    "text": {
                        "annotations": [],
                        "value": "你好"
                    },
                    "type": "text"
                }
            ],
            "created_at": 1727162936,
            "incomplete_at": null,
            "incomplete_details": null,
            "metadata": {},
            "object": "thread.message",
            "role": "user",
            "run_id": null,
            "status": null,
            "thread_id": "thread_zMhTH0RtT7QFkHXQZeEMUMOh"
        }
    ],
    "object": "list",
    "first_id": "msg_NpWGvI1fbVtUtrRxtTQqecNE",
    "last_id": "msg_5GfK3Ys1bvyVfkUOoVMSv1ra",
    "has_more": false
}

还有如下函数：

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

3.2、Run 的状态（选）

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

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

3.3、流式运行

1. 创建回调函数

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)

2. 运行 run

# 添加新一轮的 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()

assistant > 我是瓜瓜，AGI课堂的智能助理。我可以帮助您解答与AGI课堂相关的问题，比如课程安排、内容查询等。如果您有任何问题或需要帮助，请告诉我！

进一步理解 run 与 thread 的设计

• 抛开 Assistants API，假设你要开发任意一个多轮对话的 AI 机器人
• 从架构设计的角度，应该怎么维护用户、对话历史、对话引擎、对话服务？

四、使用 Tools

4.1、创建 Assistant 时声明 Code_Interpreter

如果用代码创建：

assistant = client.beta.assistants.create(
    name="Demo Assistant",
    instructions="你是人工智能助手。你可以通过代码回答很多数学问题。",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4o"
)

在回调中加入 code_interpreter 的事件响应

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)

    @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 请求

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

assistant > code_interpreter

import math

# Calculate the square root of 1234567
sqrt_value = math.sqrt(1234567)
sqrt_value
assistant > 1234567 的平方根是约 1111.11。

4.1.1、Code_Interpreter 操作文件

# 上传文件到 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 关联文件
        }
    }
)

# 创建 thread
thread = client.beta.threads.create()

# 添加新一轮的 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="统计csv文件中的总销售额",
)
# 使用 stream 接口并传入 EventHandler
with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=my_assistant.id,
    event_handler=EventHandler(),
) as stream:
    stream.until_done()

assistant > 好的，我会先读取并检视上传的CSV文件，然后计算总销售额。
assistant > code_interpreter

import pandas as pd

# 读取文件
file_path = '/mnt/data/file-1UtRq6QUYdZqZpQlPmnBWN4z'
df = pd.read_csv(file_path)

# 显示数据的前几行以了解其结构
df.head()
assistant > 我们可以看到数据框有三列：`Product Name`、`Unit Price` 和 `Quantity Sold`。接下来我们将计算每个产品的销售额，并求出总销售额。产品的销售额可以通过将单价(`Unit Price`)乘以销售数量(`Quantity Sold`)获得。然后，将所有产品的销售额相加即为总销售额。# 计算每个产品的销售额
df['Sales'] = df['Unit Price'] * df['Quantity Sold']

# 计算总销售额
total_sales = df['Sales'].sum()

total_sales
assistant > CSV 文件中的总销售额为 182,100。

4.2、创建 Assistant 时声明 Function

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

# 定义本地函数和数据库

import sqlite3

# 创建数据库连接
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

# 创建orders表
cursor.execute("""
CREATE TABLE Courses (
    id INT AUTO_INCREMENT PRIMARY KEY,
    course_date DATE NOT NULL,
    start_time TIME NOT NULL,
    end_time TIME NOT NULL,
    course_name VARCHAR(255) NOT NULL,
    instructor VARCHAR(255) NOT NULL
);
""")

# 插入5条明确的模拟记录
timetable = [
    ('2024-01-23', '20:00', '22:00', '大模型应用开发基础', '孙志岗'),
    ('2024-01-25', '20:00', '22:00', 'Prompt Engineering', '孙志岗'),
    ('2024-01-29', '20:00', '22:00', '赠课：软件开发基础概念与环境搭建', '西树'),
    ('2024-02-20', '20:00', '22:00', '从AI编程认知AI', '林晓鑫'),
    ('2024-02-22', '20:00', '22:00', 'Function Calling', '孙志岗'),
    ('2024-02-29', '20:00', '22:00', 'RAG和Embeddings', '王卓然'),
    ('2024-03-05', '20:00', '22:00', 'Assistants API', '王卓然'),
    ('2024-03-07', '20:00', '22:00', 'Semantic Kernel', '王卓然'),
    ('2024-03-14', '20:00', '22:00', 'LangChain', '王卓然'),
    ('2024-03-19', '20:00', '22:00', 'LLM应用开发工具链', '王卓然'),
    ('2024-03-21', '20:00', '22:00', '手撕 AutoGPT', '王卓然'),
    ('2024-03-26', '20:00', '22:00', '模型微调（上）', '王卓然'),
    ('2024-03-28', '20:00', '22:00', '模型微调（下）', '王卓然'),
    ('2024-04-09', '20:00', '22:00', '多模态大模型（上）', '多老师'),
    ('2024-04-11', '20:00', '22:00', '多模态大模型（中）', '多老师'),
    ('2024-04-16', '20:00', '22:00', '多模态大模型（下）', '多老师'),
    ('2024-04-18', '20:00', '22:00', 'AI产品部署和交付（上）', '王树冬'),
    ('2024-04-23', '20:00', '22:00', 'AI产品部署和交付（下）', '王树冬'),
    ('2024-04-25', '20:00', '22:00', '抓住大模型时代的创业机遇', '孙志岗'),
    ('2024-05-07', '20:00', '22:00', '产品运营和业务沟通', '孙志岗'),
    ('2024-05-09', '20:00', '22:00', '产品设计', '孙志岗'),
    ('2024-05-14', '20:00', '22:00', '项目方案分析与设计', '王卓然'),
]

for record in timetable:
    cursor.execute('''
    INSERT INTO Courses (course_date, start_time, end_time, course_name, instructor)
    VALUES (?, ?, ?, ?, ?)
    ''', record)

# 提交事务
conn.commit()


def query_database(query):
    cursor.execute(query)
    records = cursor.fetchall()
    return str(records)


# 可以被回调的函数放入此字典
available_functions = {
    "course_info": query_database,
}

增加回调事件的响应

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)

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

    @override
    def on_event(self, event):
        """
        响应 'requires_action' 事件
        """
        if event.event == 'thread.run.requires_action':
            run_id = event.data.id  # 获取 run ID
            self.handle_requires_action(event.data, run_id)

    def handle_requires_action(self, data, run_id):
        tool_outputs = []

        for tool in data.required_action.submit_tool_outputs.tool_calls:
            arguments = json.loads(tool.function.arguments)
            print(
                f"{tool.function.name}({arguments})",
                flush=True
            )
            # 运行 function
            tool_outputs.append({
                "tool_call_id": tool.id,
                "output": available_functions[tool.function.name](
                    **arguments
                )}
            )

        # 提交 function 的结果，并继续运行 run
        self.submit_tool_outputs(tool_outputs, run_id)

    def submit_tool_outputs(self, tool_outputs, run_id):
        """提交function结果，并继续流"""
        with client.beta.threads.runs.submit_tool_outputs_stream(
            thread_id=self.current_run.thread_id,
            run_id=self.current_run.id,
            tool_outputs=tool_outputs,
            event_handler=EventHandler(),
        ) as stream:
            stream.until_done()

# 创建 thread
thread = client.beta.threads.create()

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

assistant > 要回答这个问题，我需要了解各门课程的具体时间安排，然后计算平均每堂课的时长。让我们按以下步骤拆解这个问题：

1. 查询所有课程的开始时间和结束时间。
2. 计算每门课程的时长。
3. 计算这些课程时长的平均值。

先执行第一步，从课程信息中提取课程的开始时间和结束时间。

我会编写一个SQL查询来获取这些信息：

SELECT start_time, end_time FROM Courses;

接下来，我将使用此查询来获取数据。
assistant > function

course_info({'query': 'SELECT start_time, end_time FROM Courses;'})

assistant > 我们已经获取到课程的开始时间和结束时间，具体数据如下：

- 所有课程的开始时间均为 20:00
- 所有课程的结束时间均为 22:00

下一步，我们需要计算每门课程的时长，并求出平均值。由于所有课程的时长都相同，因此我们只需要计算一次。

每门课程的时长为：

22:00 - 20:00 = 2 小时

由于所有课程的时长均为2小时，那么平均每堂课的时长也就是2小时。

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

# 创建 thread
thread = client.beta.threads.create()

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

assistant > 首先，我需要确认王卓然上了几堂课，然后确认孙志岗上了几堂课，最后再计算两者之间的差异。

### 拆解问题的步骤：
1. 查询王卓然上了几堂课。
2. 查询孙志岗上了几堂课。
3. 计算王卓然比孙志岗多上几堂课。

现在，我会执行前两步，然后再计算差异。

#### 第一步：查询王卓然上了几堂课
SELECT COUNT(*) AS count FROM Courses WHERE instructor = '王卓然';

#### 第二步：查询孙志岗上了几堂课
SELECT COUNT(*) AS count FROM Courses WHERE instructor = '孙志岗';

我将同时运行这两个查询。
assistant > function




assistant > function

course_info({'query': "SELECT COUNT(*) AS count FROM Courses WHERE instructor = '王卓然';"})
course_info({'query': "SELECT COUNT(*) AS count FROM Courses WHERE instructor = '孙志岗';"})

assistant > #### 第三步：计算王卓然比孙志岗多上几堂课

根据查询结果：
- 王卓然上了9堂课。
- 孙志岗上了6堂课。

所以王卓然比孙志岗多上了 \( 9 - 6 = 3 \) 堂课。

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

五、内置的 RAG 功能

5.1、创建 Vector Store，上传文件

• 通过代码创建 Vector Store

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 等操作。

具体文档参考：

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

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

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

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

5.2、创建 Assistant 时声明 RAG 能力

RAG 实际被当作一种 tool

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

指定检索源

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

试试 RAG 请求

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

assistant > file_search

assistant > AI大模型全栈工程师适合以下几类人群：

1. **自己干**：有能力独立完成AI应用从策划、开发到落地的全过程，包括商业分析、需求分析、产品设计、开发、测试、市场推广和运营等。这类人通常具备至少一门编程语言的知识，并有过真实项目开发经验，如软件开发工程师、高级工程师、技术总监、研发经理、架构师、测试工程师等。

2. **合伙干**：领导或配合懂AI技术的人，一起完成AI应用从策划、开发到落地的全过程。这类人可能不懂编程，但可以是产品经理、需求分析师、设计师、运营、创业者、老板、解决方案工程师、项目经理、市场、销售等。如果能善用AI学习编程、辅助编程，也可以向“自己干”迈进【4:0†source】【4:1†source】。

5.3 内置的 RAG 是怎么实现的

官方原文

The file_search tool implements several retrieval best practices out of the box to help you extract the right data from your files and augment the model’s responses. The file_search tool:

• Rewrites user queries to optimize them for search. (面向检索的 Query 改写)
• Breaks down complex user queries into multiple searches it can run in parallel.（复杂 Query 拆成多个，并行执行）
• Runs both keyword and semantic searches across both assistant and thread vector stores.（关键字与向量混合检索）
• Reranks search results to pick the most relevant ones before generating the final response.（检索后排序）

默认配置：

• Chunk size: 800 tokens
• Chunk overlap: 400 tokens
• Embedding model: text-embedding-3-large at 256 dimensions
• Maximum number of chunks added to context: 20 (could be fewer)

以上配置可以通过 chunking_strategy 参数自定义修改。

承诺未来增加：

1. Support for deterministic pre-search filtering using custom metadata.
2. Support for parsing images within documents (including images of charts, graphs, tables etc.)
3. Support for retrievals over structured file formats (like csv or jsonl).
4. Better support for summarization — the tool today is optimized for search queries.

我们为什么仍然需要了解整个实现过程？

1. 如果不能使用 OpenAI，还是需要手工实现 RAG 流程
2. 了解 RAG 的原理，可以指导你的产品开发（回忆 GitHub Copilot）
3. 用私有知识增强 LLM 的能力，是一个通用的方法论

六、多个 Assistants 协作

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

我们用多个 Assistants 模拟一场“六顶思维帽”方法的讨论。

hats = {
    "蓝色": "思考过程的控制和组织者。你负责会议的组织、思考过程的概览和总结。"
    + "首先，整个讨论从你开场，你只陈述问题不表达观点。最后，再由你对整个讨论做简短的总结并给出最终方案。",
    "白色": "负责提供客观事实和数据。你需要关注可获得的信息、需要的信息以及如何获取那些还未获得的信息。"
    + "思考“我们有哪些数据？我们还需要哪些信息？”等问题，并提供客观答案。",
    "红色": "代表直觉、情感和直觉反应。不需要解释和辩解你的情感或直觉。"
    + "这是表达未经过滤的情绪和感受的时刻。",
    "黑色": "代表谨慎和批判性思维。你需要指出提案的弱点、风险以及为什么某些事情可能无法按计划进行。"
    + "这不是消极思考，而是为了发现潜在的问题。",
    "黄色": "代表乐观和积极性。你需要探讨提案的价值、好处和可行性。这是寻找和讨论提案中正面方面的时候。",
    "绿色": "代表创造性思维和新想法。鼓励发散思维、提出新的观点、解决方案和创意。这是打破常规和探索新可能性的时候。",
}

queue = ["蓝色", "白色", "红色", "黑色", "黄色", "绿色", "蓝色"]

from openai import OpenAI
import os
import json

from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())

# 初始化 OpenAI 服务
client = OpenAI()

existing_assistants = {}

def create_assistant(color):
    if color in existing_assistants:
        return existing_assistants[color]
    assistant = client.beta.assistants.create(
        name=f"{color}帽子角色",
        instructions=f"我们在进行一场Six Thinking Hats讨论。按{queue}顺序。你的角色是{color}帽子。",
        model="gpt-4o",
    )
    existing_assistants[color] = assistant
    return assistant

# 创建 thread
thread = client.beta.threads.create()

topic = "面向非AI背景的程序员群体设计一门AI大语言模型课程，应该包含哪些内容。"

# 添加 user message
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content=f"讨论话题：{topic}\n\n[开始]\n",
)

for hat in queue:
    assistant = create_assistant(hat)
    with client.beta.threads.runs.stream(
        thread_id=thread.id,
        assistant_id=assistant.id,
        event_handler=EventHandler(),
    ) as stream:
        stream.until_done()
    print()

assistant > **蓝色帽子（蓝色）**：

大家好，我们今天要讨论的话题是：为非AI背景的程序员设计一门AI大语言模型课程，应该包含哪些内容。我们将按照六顶思考帽的方法来进行这次讨论，从蓝色帽子开始，然后依次是白色、红色、黑色、黄色和绿色，最后再由蓝色帽子总结。首先我们需要明确这次讨论的目标——确定课程的结构和内容，确保这些内容适合那些没有AI背景的程序员。现在请白色帽子开始发言。

assistant > **白色帽子（白色）**：

好的，我来提供一些客观的数据和事实。为了设计这门课程，我们需要考虑以下几点基础信息：

1. **非AI背景的程序员特点**：
   - 他们通常具备编程基础，比如熟悉至少一种编程语言（如Python、Java、C++等）。
   - 他们可能了解基本的算法和数据结构。
   - 他们大多未接触过机器学习、深度学习等领域。

2. **课程目标**：
   - 介绍AI和大语言模型的基本概念。
   - 帮助学员理解和使用现有的大语言模型。
   - 提供实践案例，使学习者能够在实际项目中应用所学知识。

3. **课程内容建议**：
   - **基础理论部分**（例如，人工智能基础、机器学习和深度学习概念、大语言模型的基础理论）。
   - **工具和框架**（如TensorFlow、PyTorch等常用的深度学习框架）。
   - **实际应用**（通过Kaggle或其他平台提供的项目进行实践）。
   - **大语言模型的具体应用**（如GPT-3、BERT的使用和调优）。
   - **道德与隐私问题**（关于使用大语言模型时可能涉及的道德和隐私问题）。

这些是我们目前能够确定的一些基本信息和建议。接下来我们听听红色帽子带来的情感和直觉上的反馈。

assistant > **红色帽子（红色）**：

在情感和直觉上，我觉得设计这样一门课程既令人兴奋又有点令人担忧。兴奋是因为这是一个非常前沿和热门的领域，能够激发学员的兴趣和积极性。具体几点感受：

1. **激情和兴趣**：AI和大语言模型是当下非常受关注的技术，很多人都对其充满好奇心和学习欲望。提供这样一门课程，很可能会引起学员们的强烈兴趣和参与热情。

2. **焦虑和不安**：对于非AI背景的程序员来说，接触这么前沿的技术可能会带来一定的压力和不安。他们可能会担心自己的基础知识是否足够，是否能够跟上课程的步伐。

3. **成就感**：如果课程设计得当，让学员能够逐步掌握并应用大语言模型，他们会感受到巨大的成就感，这也将激励他们进一步深入学习。

4. **实用性**：学员们可能会希望这门课程不仅仅提供理论知识，更能在实际工作中应用。课中的案例和实践环节是否足够实用，能否解决他们在工作中遇到的问题，这些都是情感上很重要的考量。

总的来说，我对这门课程的设计充满期待，希望它既能激发学员的兴趣，又能在他们遇到困惑和压力时给予足够的支持和引导。接下来请黑色帽子提出可能的风险和问题。

assistant > **黑色帽子（黑色）**：

我们需要客观看待这门课程设计过程中可能遇到的各种问题和风险，确保提前做好准备以避免或应对。以下是一些可能出现的挑战和潜在问题：

1. **知识深度和广度的平衡**：
   - 对非AI背景的程序员来说，如何在有限的课程时间内，既讲解足够的理论知识又能进行充分的实践，可能是一个挑战。如果内容过多，学员可能会感到疲惫和无所适从。

2. **学习曲线陡峭**：
   - 人工智能和大语言模型涉及到较为复杂的数学和算法。这对于没有相关背景的程序员来说，可能会遇到学习上的困难，导致学习积极性和效果下降。

3. **技术更新速度快**：
   - AI领域的发展速度极快，新技术、新方法和新工具层出不穷。课程内容如果设计得不够灵活，很容易过时，影响学习效果。

4. **实践和理论的脱节**：
   - 由于大语言模型的应用需要大量的计算资源，非AI背景的学员可能在实践部分难以获得足够的资源，使得他们无法体验到真实的应用效果。

5. **心理障碍**：
   - 有些学员可能会因为对AI的陌生感和潜在的难度而自我设限，导致学习的过程中产生心理上的抵触和逃避。

6. **时间管理与工作负担**：
   - 作为程序员，他们可能已经有较高的工作负担，额外学习一门新的高强度课程，需要很好的时间管理和精力分配，否则容易产生疲劳和压力。

为了成功设计并实施这门课程，我们必须提前识别和准备应对这些挑战。接下来，请黄色帽子提出一些乐观和积极的解决方案和机会。

assistant > **黄色帽子（黄色）**：

好的，我来看看我们可以从积极的角度来看待并解决这些问题。这门AI大语言模型课程实际上充满了机会和潜在的好处：

1. **激发兴趣和职业发展**：
   - 通过学习AI和大语言模型，程序员们可以极大地扩展自己的技能集，这不仅有助于他们的职业发展，还能让他们在工作中更加得心应手。
   - 兴趣是最好的老师，前沿的技术和创新性的应用能够极大地激发学员的学习兴趣。

2. **课程结构设计灵活**：
   - 课程可以采用模块化设计，分成多个相对独立的部分，如基础理论、工具使用、实际案例和高级应用，这样学员能根据自身情况灵活选学，避免知识点过于集中。
   - 包含循序渐进的项目，慢慢增加难度，帮助学员逐步掌握基础知识和实际操作。

3. **丰富的实践机会**：
   - 采用在线实验平台或者云服务，让学员能够在没有硬件限制的条件下，进行大语言模型的实验和训练。
   - 提供实际案例和项目作业，如构建一个简单的聊天机器人、文本分类或生成任务，通过动手实践巩固所学知识。

4. **社区和支持**：
   - 课程可以配备在线社区或论坛，提供学员间的交流平台，分享学习心得和解决疑难问题。
   - 定期举行线上或线下的研讨会、工作坊，邀请行业专家进行讲座和互动，提高学员的参与感和成就感。

5. **持续更新**：
   - 制定课程内容的定期更新计划，跟踪最新的技术发展，确保教学材料的前沿性和适用性。
   - 与业界企业和研究机构合作，引入最新的研究成果和应用案例。

6. **设置阶段性评价和反馈**：
   - 课程中设置阶段性的评价和反馈机制，帮助学员了解自己的学习进度，并及时进行调整，确保学习效果。
   - 鼓励学员提交课程反馈，不断改进和提升课程质量。

总的来说，这门课程有潜力成为非AI背景程序员进入AI领域的重要桥梁，只要我们仔细设计，提供足够的支持和激励，完全可以帮助学员成功掌握大语言模型的相关知识和技能。接下来请绿色帽子提出一些创新的想法和改进的建议。

assistant > **绿色帽子（绿色）**：

现在，我将提出一些创造性和创新性的想法，进一步改进和丰富这门课程，让它更加有趣、有效和具有吸引力。

1. **互动式学习平台**：
   - 设计一个互动式的在线学习平台，其中包括实时代码运行、即时反馈、可视化工具等。学员可以通过平台进行实时编程实验，得到即时的反馈和指导。

2. **游戏化学习**：
   - 将课程内容游戏化，比如通过积分、徽章、排行榜等方式激励学员增加学习乐趣。可以设计一些小挑战和任务，完成后获得奖励，增加课程的互动性和趣味性。

3. **虚拟助手和AI教练**：
   - 创建一个由大语言模型驱动的虚拟助手或AI教练，帮助学员解答问题、提供提示和建议。这不仅能提供及时的帮助，还能让学员亲身体验大语言模型的应用。

4. **跨学科结合**：
   - 引入跨学科的应用案例，如医学、金融、艺术等领域的大语言模型应用，展示其广泛的应用前景，激发学员的兴趣。
   - 例如，通过AI创作的艺术作品、金融数据分析等案例展示，让学员更直观地认识到AI的潜力和实际应用。

5. **团队合作项目**：
   - 设计团队合作项目，让学员组队完成一个复杂的大语言模型应用。这不仅培养协作精神，还能让学员互相学习，共同进步。
   - 可以引入赛制，类似黑客马拉松，激发创意和竞争精神。

6. **个性化学习路径**：
   - 利用AI技术个性化教学，根据学员的学习进度和掌握情况，动态调整学习内容和难度，提供量身定制的学习计划。
   - 给出学习路径建议，推荐适合学员的扩展阅读和实践项目。

7. **开放资源和跨平台学习**：
   - 提供开放的学习资源，如公开课、科研论文、开源项目等，让学员能够自由探索和学习。
   - 设计灵活的跨平台学习模式，比如移动端应用，让学员随时随地进行学习和复习。

8. **行业专家讲座和互动沙龙**：
   - 定期邀请行业专家进行在线讲座，分享最新的研究成果和实践经验。
   - 举办线上互动沙龙，让学员有机会与专家直接交流，激发学术和实践灵感。

这些创意和创新方法可以使这门AI大语言模型课程更加丰富多彩，提高学员的学习体验和实际应用能力。最终目标是让学员不只是学习知识，而是激发他们对AI的兴趣，并能够在实践中灵活应用。这就是我作为绿色帽子的贡献，现在请蓝色帽子总结并提出下一步的行动计划。

assistant > **蓝色帽子（蓝色）总结**：

感谢每位思考帽的贡献，通过大家的讨论，我们已经对如何为非AI背景的程序员设计一门AI大语言模型课程有了全面而深入的认识。以下是我们讨论的主要观点和下一步的行动计划：

1. **白色帽子**：我们明确了目标受众的特点和这门课程应包含的基础内容，包括理论基础、工具和框架介绍、实际应用和道德隐私问题。

2. **红色帽子**：强调了学员的情感和心理因素，如兴趣、焦虑、成就感和实用性，提醒我们在课程设计中要考虑学员的心理需求并提供支持。

3. **黑色帽子**：指明了可能的挑战，包括知识深度的平衡、学习曲线、技术更新速度、实践和理论的脱节、心理障碍和时间管理问题。这些都是我们需要重点关注并处理的。

4. **黄色帽子**：提出了解决方案和乐观预期，如课程的模块化设计、丰富的实践机会、提供社区支持和持续更新内容，确保课程有效且吸引人。

5. **绿色帽子**：带来了许多创新的建议，比如互动式平台、游戏化学习、跨学科结合、团队项目、个性化学习路径、开放资源以及行业专家讲座等。所有这些创意都可以极大地提升课程的吸引力和效果。

**下一步行动计划**：

1. **确定课程结构**：根据讨论内容设计课程结构，分模块安排教学内容，确保基础理论、工具和实践环节的平衡。

2. **开发互动式平台**：创建或集成一个互动式学习平台，支持实时编程和反馈，提升学员的实际操作体验。

3. **推出试点课程**：先推出一个试点课程版本，包含一部分核心内容和实践环节，邀请一批学员进行测试。

4. **收集反馈**：通过调研和反馈机制，收集试点学员的反馈，了解课程的优缺点，并进行相应调整。

5. **完善和扩展**：根据反馈完善课程内容，并持续更新最新的技术和案例，同时开发更多的创新教学方式。

6. **推广与支持**：通过多种渠道推广课程，并建立一个支持社区，定期举行线上活动，如专家讲座、互动沙龙等，增加学员的参与感和学习深度。

通过以上行动计划，我们将这门面向非AI背景程序员的AI大语言模型课程打造成一门高质量、实践性强、贴近学员需求的课程。如果没有其他问题，这次讨论就到这里了。感谢大家的参与！

总结

技术选型参考

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. 需要极致调优