English
主导航

推荐

入门

核心概念

Apps SDK

工具

运行与扩展

评估

实时与音频

模型优化

专业模型

正式上线

旧版 API

资源

API 仪表板

通过 ChatKit 实现高级集成

将您自己的基础设施与 ChatKit 搭配使用，以实现更多自定义。

当您需要完全控制时——无论是自定义身份验证、数据驻留、本地部署，还是定制化的智能体编排——您都可以在您自己的基础设施上运行 ChatKit。使用 OpenAI 的高级自托管选项，以运行您自己的服务器和自定义的 ChatKit。

我们推荐的 ChatKit 集成可帮助您快速入门：嵌入聊天小部件，自定义其外观和风格，并由 OpenAI 负责托管和扩展后端。 使用更简单的集成 →

在您自己的基础设施上运行 ChatKit

概括而言，高级 ChatKit 集成是一个构建您自己的 ChatKit 服务器并添加小部件以构建聊天界面的过程。您将使用 OpenAI API 和您的 ChatKit 服务器，构建一个由 OpenAI 模型驱动的自定义聊天功能。

OpenAI-hosted ChatKit

设置您的 ChatKit 服务器

遵循 GitHub 上的服务器指南 以了解如何处理传入请求、运行工具以及将结果流式传输回客户端。以下代码片段重点介绍了主要组件。

1. 安装服务器包

pip install openai-chatkit

2. 实现服务器类

ChatKitServer 驱动对话。重写 respond 以便在用户消息或客户端工具输出到达时流式传输事件。诸如 stream_agent_response 之类的辅助工具使得与 Agents SDK 的连接变得简单。

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
class MyChatKitServer(ChatKitServer):
    def __init__(self, data_store: Store, file_store: FileStore | None = None):
        super().__init__(data_store, file_store)

    assistant_agent = Agent[AgentContext](
        model="gpt-4.1",
        name="Assistant",
        instructions="You are a helpful assistant",
    )

    async def respond(
        self,
        thread: ThreadMetadata,
        input: UserMessageItem | ClientToolCallOutputItem,
        context: Any,
    ) -> AsyncIterator[Event]:
        agent_context = AgentContext(
            thread=thread,
            store=self.store,
            request_context=context,
        )
        result = Runner.run_streamed(
            self.assistant_agent,
            await to_input_item(input, self.to_message_content),
            context=agent_context,
        )
        async for event in stream_agent_response(agent_context, result):
            yield event

    async def to_message_content(
        self, input: FilePart | ImagePart
    ) -> ResponseInputContentParam:
        raise NotImplementedError()

3. 暴露端点

使用您选择的框架将 HTTP 请求转发到服务器实例。例如，使用 FastAPI：

1
2
3
4
5
6
7
8
9
10
11
app = FastAPI()
data_store = SQLiteStore()
file_store = DiskFileStore(data_store)
server = MyChatKitServer(data_store, file_store)

@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
    result = await server.process(await request.body(), {})
    if isinstance(result, StreamingResult):
        return StreamingResponse(result, media_type="text/event-stream")
    return Response(content=result.json, media_type="application/json")

4. 建立数据存储契约

实现 chatkit.store.Store 以使用您首选的数据库来持久化线程、消息和文件。默认示例使用 SQLite 进行本地开发。考虑将模型存储为 JSON blob，以便库更新时无需迁移即可改进架构。

5. 提供文件存储契约

如果您支持上传功能，请提供一个 FileStore 实现。ChatKit 支持直接上传（客户端将文件 POST 到您的端点）或两阶段上传（客户端请求一个带签名的 URL，然后上传到云存储）。暴露预览以支持内联缩略图，并在移除线程时处理删除操作。

6. 从服务器触发客户端工具

客户端工具必须在客户端选项和您的智能体上同时注册。使用 ctx.context.client_tool_call 从 Agents SDK 工具中将调用排入队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@function_tool(description_override="Add an item to the user's todo list.")
async def add_to_todo_list(ctx: RunContextWrapper[AgentContext], item: str) -> None:
    ctx.context.client_tool_call = ClientToolCall(
        name="add_to_todo_list",
        arguments={"item": item},
    )

assistant_agent = Agent[AgentContext](
    model="gpt-4.1",
    name="Assistant",
    instructions="You are a helpful assistant",
    tools=[add_to_todo_list],
    tool_use_behavior=StopAtTools(stop_at_tool_names=[add_to_todo_list.name]),
)

7. 使用线程元数据和状态

使用 thread.metadata 来存储服务端状态，例如之前的 Responses API 运行 ID 或自定义标签。元数据不会向客户端公开，但在每次 respond call.

8. 获取工具状态更新

长时间运行的工具可以通过以下方式将进度流式传输到 UI： ProgressUpdateEvent. ChatKit 会用下一条助手消息或组件输出替换进度事件。

9. 使用服务器上下文

将自定义上下文对象传递给 server.process(body, context) 以在您的存储和文件存储实现中强制执行权限或传播用户身份。

添加内联交互式组件

组件让代理能在聊天界面中呈现富 UI。可将它们用于卡片、表单、文本块、列表及其他布局。该辅助工具 stream_widget 可以立即渲染组件，或在数据到达时流式传输更新。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def respond(
    self,
    thread: ThreadMetadata,
    input: UserMessageItem | ClientToolCallOutputItem,
    context: Any,
) -> AsyncIterator[Event]:
    widget = Card(
        children=[Text(
            id="description",
            value="Generated summary",
        )]
    )
    async for event in stream_widget(
        thread,
        widget,
        generate_id=lambda item_type: self.store.generate_item_id(item_type, thread, context),
    ):
        yield event

ChatKit 提供了一系列丰富的组件节点（卡片、列表、表单、文本、按钮等）。请参阅 GitHub 上的组件指南 以了解所有组件、属性和流式传输指南。

参阅 GitHub 上的 组件构建器 以在交互式 UI 中探索和创建组件。

使用操作

操作允许 ChatKit UI 在不发送用户消息的情况下触发任务。将 ActionConfig 附加到任何支持它的组件节点上——按钮、选择框及其他控件可以流式传输新的对话项或就地更新组件。当组件位于 Form内时，ChatKit 会将收集到的表单值包含在操作负载中。

在服务器端，实现 action 方法与 ChatKitServer 以处理负载并可选地流式传输额外事件。你也可以通过设置 handler="client" 并在 JavaScript 中进行响应，然后再将后续任务转发给服务器，从而在客户端处理操作。

参阅 GitHub 上的 GitHub 上的操作指南 以了解链式操作、创建强类型负载以及协调客户端/服务器处理程序等模式。

资源

使用以下资源和参考来完成你的集成。

设计资源

事件参考

ChatKit 发出 CustomEvent 来自 Web Component 的实例。负载结构为：

1
2
3
4
5
6
7
type Events = {
  "chatkit.error": CustomEvent<{ error: Error }>;
  "chatkit.response.start": CustomEvent<void>;
  "chatkit.response.end": CustomEvent<void>;
  "chatkit.thread.change": CustomEvent<{ threadId: string | null }>;
  "chatkit.log": CustomEvent<{ name: string; data?: Record<string, unknown> }>;
};

选项参考

选项类型描述默认值
apiURLstring实现 ChatKit 服务器协议的端点。必需
fetchtypeof fetch覆盖 fetch 调用（用于自定义请求头或身份验证）。window.fetch
theme"light" | "dark"UI 主题。"light"
initialThreadstring | null挂载时打开的线程； null 显示新线程视图。null
clientToolsRecord<string, Function>向模型暴露的由客户端执行的工具。
headerobject | boolean头部配置或 false to hide the header.true
newThreadViewobject自定义问候语和起始提示。
messagesobject配置消息附加功能（反馈、注释等）。
composerobject控制附件、实体标签和占位符文本。
entitiesobject用于实体查找、点击处理和预览的回调函数。