大模型应用开发：使用 OpenAI Assistants 构建产品 AI 客服

大模型应用开发：使用 OpenAI Assistants 构建产品 AI 客服 | 极客日志

大模型应用开发：使用 OpenAI Assistants 构建产品 AI 客服

引言

在大型语言模型（LLM）的应用开发中，如何赋予模型特定的领域知识并使其稳定地服务于业务场景是一个核心挑战。OpenAI 推出的 Assistants API 提供了一种标准化的方式来创建具有特定指令、知识库和工具能力的智能助手。本文将详细介绍如何使用 Python SDK 结合 OpenAI Assistants API，为一款空气净化器产品构建一个基于产品手册的 AI 客服系统。

什么是 Assistants？

在之前的开发实践中，我们可能使用过插件（Plugins）来扩展大模型的能力，例如查询实时天气或进行数学运算。插件主要解决了大模型自身无法获取外部数据或执行特定计算的问题。

Assistants 则更进一步，它旨在强化大模型在特定应用场景下的综合能力。通过 Assistants，我们可以实现以下功能：

精准的知识问答：如 AI 客服和知识库助手，能够准确理解用户问题，并在限定的文档范围内提供精准回答，减少幻觉。
代码审查与生成：按照指定的规范对代码进行 Review，或按照某种风格进行文学创作。
复杂任务编排：结合代码解释器和函数调用，处理多步骤的任务流程。

本文将以一个空气净化器 AI 客服为例，演示 Assistants 的核心用法。该客服将严格依据上传的产品手册回答问题，对于超出范围的问题会礼貌拒绝。

Assistants 的运行原理

在编写代码之前，理解 Assistants 的工作流至关重要。其核心流程包含五个关键步骤：

创建智能助手 (Assistant)：定义助手的身份，包括名称、描述、使用的模型版本、系统指令（Instructions）以及启用的工具（Tools）。
创建用户会话 (Thread)：会话是用户与助手之间的一次对话上下文。GPT 通过 Thread 管理聊天的历史状态。
添加用户消息 (Message)：将用户的输入添加到会话中。每条消息都有角色标识（user 或 assistant）。
运行智能助手 (Run)：将线程和助手绑定，触发助手的推理过程。这一步会创建一个执行对象，并将其加入处理队列。
获取响应消息 (Response)：轮询运行状态，当状态变为完成时，从线程中提取助手的回复。

环境准备

在开始编码前，请确保已完成以下准备工作：

安装 OpenAI Python SDK：
```
pip install openai
```

配置 API Key：建议将 API Key 存储在环境变量中，避免硬编码在代码里。

import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

准备知识库文件：准备一份产品手册（如 TXT 或 PDF 格式），内容应清晰结构化，便于模型检索。

实现 AI 客服

1. 上传产品手册

首先，需要将产品手册上传到 OpenAI 的文件存储中。注意设置 purpose 为 assistants，以便在检索工具中使用。

from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 上传文件
file_response = client.files.create(
    file=open("product_manual.txt", "rb"),
    purpose="assistants"
)
file_id = file_response.id
print(f"File uploaded with ID: {file_id}")

2. 创建智能助手

使用 client.beta.assistants.create 创建客服实例。由于 Assistants API 处于 Beta 阶段，包名中包含 beta。

assistant = client.beta.assistants.create(
    name="空气净化器智能客服",
    description="24 小时为您服务，基于产品手册解答疑问",
    instructions="你是空气净化器公司的智能客服。请引用文件中的内容回答问题，表达要通俗易懂、尽量简短；若问题超出文件内容，请委婉拒绝。",
    model="gpt-3.5-turbo-1106",
    tools=[{"type": "retrieval"}],
    file_ids=[file_id]
)
assistant_id = assistant.id
print(f"Assistant created with ID: {assistant_id}")

参数说明：

name：助手的显示名称。
instructions：系统提示词，定义助手的角色和行为准则。这是控制输出质量的关键。
model：指定使用的模型版本。目前推荐使用最新的 Turbo 版本以平衡成本与性能。
tools：启用检索能力 (retrieval)，使助手能读取上传的文件。
file_ids：关联已上传的知识库文件 ID。

3. 创建用户会话

每个用户的对话都需要独立的会话线程来维护上下文。

thread = client.beta.threads.create(
    metadata={
        "user_name": "张三",
        "user_id": "12345"
    }
)
thread_id = thread.id

4. 添加用户消息

将用户的问题添加到会话中。

message = client.beta.threads.messages.create(
    thread_id=thread_id,
    role="user",
    content="这款净化器支持哪些模式？"
)

5. 运行智能助手

触发助手处理当前会话的消息。

run = client.beta.threads.runs.create(
    thread_id=thread_id,
    assistant_id=assistant_id
)
run_id = run.id

6. 获取智能助手的回应

由于 Run 是异步处理的，我们需要轮询其状态直到完成。生产环境中建议使用指数退避策略（Exponential Backoff）以避免频繁请求。

import time

while run.status in ["queued", "in_progress"]:
    time.sleep(1) # 实际生产中建议增加随机延迟
    run = client.beta.threads.runs.retrieve(
        thread_id=thread_id,
        run_id=run_id
    )

if run.status == "completed":
    messages = client.beta.threads.messages.list(
        thread_id=thread_id,
        order="asc"
    )
    # 获取最后一条助手回复
    assistant_message = next(m for m in messages.data if m.role == "assistant")
    print(assistant_message.content[0].text)
elif run.status == "failed":
    print(f"Error: {run.last_error.message}")
else:
    print(f"Status: {run.status}")

状态说明：

queued：任务排队中。
in_progress：正在处理中。
completed：处理成功。
requires_action：需要调用函数（本例未涉及）。
failed：运行失败。

7. 处理引用注解 (Annotations)

当助手引用知识库内容时，返回的消息中会包含 annotations。这允许我们在前端展示脚注，让用户点击查看原文来源。

def extract_message_content(message):
    text_content = message.content[0].text
    annotations = text_content.annotations
    
    citations = []
    value = text_content.value
    
    for index, annotation in enumerate(annotations):
        # 替换文本为脚注标记
        value = value.replace(annotation.text, f' [{index}]')
        
        # 收集引用信息
        if hasattr(annotation, 'file_citation'):
            cited_file = client.files.retrieve(annotation.file_citation.file_id)
            citations.append(f'[{index}] 引用自：{cited_file.filename}')
        elif hasattr(annotation, 'file_path'):
            cited_file = client.files.retrieve(annotation.file_path.file_id)
            citations.append(f'[{index}] 下载源文件：{cited_file.filename}')
    
    return value + '\n\n'.join(citations)

最佳实践与注意事项

API 密钥安全：永远不要在前端代码或公开仓库中暴露 API Key。应在服务端通过环境变量管理。
费用控制：Retrieval 功能会增加 Token 消耗。建议定期清理不再需要的文件和 Thread。
错误处理：网络波动可能导致 failed 状态。建议实现重试机制，针对 rate_limit_exceeded 等特定错误进行指数退避。
内容合规：虽然 Retrieval 限制了回答范围，但仍需监控输出内容，防止敏感信息泄露或不当言论。
模型选择：对于复杂的逻辑推理，可考虑升级到 GPT-4 系列模型，但需注意成本差异。

总结

通过 OpenAI Assistants API，开发者可以快速构建具备领域知识的智能客服，而无需从头训练模型。本文展示了从文件上传、助手配置到会话管理和响应获取的完整流程。借助检索增强生成（RAG）技术，企业能够更可靠地将大模型能力落地到具体业务场景中，提升用户体验的同时降低运营成本。

大模型应用开发：使用 OpenAI Assistants 构建产品 AI 客服