跳到主要内容

搭建全栈应用

本教程会搭建一个真正的聊天产品：React 浏览器客户端、你自己的后端、Iztro Agent SDK，以及你自己的业务数据库。后端可以选择 Python/FastAPI 或 TypeScript/Express。

你不需要自己实现对话记忆，也不需要在每一轮重新拼装全部消息，更不能把 Iztro API Key 放进浏览器。但用户认证、权限判断、产品数据和所有业务副作用仍然由你的系统负责。

完整源码：

先理解整体架构​

React 浏览器
    |  你的 JSON API + Server-Sent Events（没有 Iztro API Key）
你的后端：FastAPI 或 Express
    |-- 用户认证与会话归属校验
    |-- ChatSession ----------> Iztro Conversations（消息上下文）
    |-- Agent 流式调用 -------> Iztro 托管模型与命盘工具
    `-- 你的数据库 -----------> 用户、套餐、标题、已保存报告、业务数据

最重要的原则是：

ChatSession 负责对话上下文；你的数据库负责业务事实与权限。

什么数据由谁保存？​

数据负责人原因
用户与 Assistant 的消息历史Iztro ChatSessionAgent 能直接恢复上下文，后端不必每次重建 Prompt。
登录、角色、订阅、额度你的后端与数据库它们决定谁能执行操作，不能从聊天文字中推断。
订单、预约、CRM、已保存报告你的数据库它们是业务记录，可能需要事务、审计和删除规则。
会话标题、置顶、反馈、分支关系通常是你的数据库它们是产品界面功能，不是模型上下文。
出生资料由你决定如果产品需要复用，应在用户同意后保存结构化资料；否则只放在对话里。
Iztro API Key后端环境变量或密钥管理器浏览器绝不能拿到密钥。
本地工具执行你的后端你的代码负责校验输入、检查权限和控制副作用。

conversation_id 只能标识一个托管对话，不能证明请求者拥有它。读取、发消息、改名、Fork 和删除时都必须校验当前登录用户。

最终会做出什么​

完整 Demo 支持：

  • 新建、列出、恢复、改名、Fork 和删除对话；
  • 将回答文字和 Iztro 命盘工具事件实时传给 React；
  • 本地只存产品元数据，不重复保存整份聊天记录；
  • 通过 external_user_id 隔离不同用户的会话；
  • API Key 始终只存在于后端。

第一次接入时，建议先完成“新建 → 发送 → 流式返回 → 恢复会话”，再加入改名、Fork 和编辑历史消息。

第 1 步：先运行完整 Demo​

你需要从 Iztro Console 获取 API Key，并安装 Node.js 20.19+（或 22.12+）。Python 版本还需要 Python 3.10+。

git clone https://github.com/SylarLong/openai-iztro-agents-python.git
cd openai-iztro-agents-python/examples/fullstack-demo/backend
python -m venv .venv
# macOS/Linux: source .venv/bin/activate
# Windows PowerShell: .venv\Scripts\Activate.ps1
pip install -r requirements.txt
cp .env.example .env

在 backend/.env 中设置 ZIWEI_API_KEY，然后启动 FastAPI：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8788

另开一个终端：

cd openai-iztro-agents-python/examples/fullstack-demo/frontend
npm install
npm run dev

打开 http://localhost:5192。

示例通过本地 editable/file 依赖直接使用克隆仓库里的 SDK 源码。你在自己的项目中应安装已发布的包：

pip install openai-iztro-agents fastapi "uvicorn[standard]"

第 2 步：在后端确认用户身份​

Demo 里的“演示用户”切换器只是为了在本地直观看到会话隔离。正式环境不能信任这种方式，因为任何人都能修改浏览器提交的 external_user_id。

生产接口应先校验 Session Cookie 或 Bearer Token，再从已验证的身份中得到稳定的用户 ID：

@app.post("/api/conversations/{conversation_id}/messages/stream")
async def stream_message(
    conversation_id: str,
    request: StreamMessageRequest,
    current_user: User = Depends(require_user),
):
    user_id = str(current_user.id)  # 绝不从请求 JSON 中读取。
    await ensure_owned(conversation_id, user_id)
    ...

对不属于当前用户的会话返回 404，不要泄露它是否存在。列出、读取、改名、Fork、编辑和删除接口都要执行相同的归属校验。

第 3 步：创建 ChatSession，只保存必要映射​

在后端用已认证的用户 ID 创建 ChatSession。托管对话会按需创建，其 ID 在 Python 中是 session_id，在 TypeScript 中是 sessionId。

from iztro_agents import ChatSession

session = ChatSession(
    external_user_id=user_id,
    api_key=os.environ["ZIWEI_API_KEY"],
)
await session.get_items()  # 如有需要，会在这里创建托管对话。
conversation_id = session.session_id

最小业务表可以这样设计：

create table agent_conversations (
  id uuid primary key,
  user_id uuid not null references users(id),
  iztro_conversation_id text not null unique,
  title text not null default '新会话',
  parent_conversation_id text,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create index agent_conversations_owner
  on agent_conversations(user_id, updated_at desc);

这张表负责权限归属和界面元数据。除非产品有独立的合规或分析需求，并且隐私政策已覆盖这份副本，否则不要把每条消息再复制一遍。

第 4 步：从后端流式运行一轮 Agent​

后端恢复托管会话、运行 Agent，再把 SDK 事件转换成一组简单、稳定、面向浏览器的 SSE 事件。

import json
import os
from agents import Runner
from fastapi.responses import StreamingResponse
from openai.types.responses import ResponseTextDeltaEvent
from iztro_agents import ChatSession, IztroToolEvent, iztro_ziwei_agent

def sse(event: str, data: object) -> bytes:
    payload = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
    return f"event: {event}\ndata: {payload}\n\n".encode()

async def events(conversation_id: str, user_id: str, message: str):
    session = ChatSession(
        conversation_id=conversation_id,
        external_user_id=user_id,
        api_key=os.environ["ZIWEI_API_KEY"],
    )
    try:
        result = Runner.run_streamed(
            iztro_ziwei_agent(api_key=os.environ["ZIWEI_API_KEY"]),
            message,
            session=session,
        )
        async for event in result.stream_events():
            if event.type != "raw_response_event":
                continue
            if isinstance(event.data, IztroToolEvent):
                yield sse("chart", {"tools": event.data.tools})
            elif isinstance(event.data, ResponseTextDeltaEvent) and event.data.delta:
                yield sse("delta", {"text": event.data.delta})
        yield sse("done", {"conversation_id": session.session_id})
    except Exception:
        yield sse("error", {"message": "回答没有完成，请重试。"})
    finally:
        await session.close()

return StreamingResponse(
    events(conversation_id, user_id, request.message),
    media_type="text/event-stream",
    headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)

完整 Demo 在元数据变化时还会发送 conversation。浏览器协议应尽量小而稳定：

事件数据浏览器行为
conversation会话摘要更新标题、分支关系和列表元数据。
chart{tools: string[]}显示实际调用了哪一种 Iztro 托管命盘工具。
delta{text: string}将文字追加到正在生成的 Assistant 消息。
done最终 ID/摘要将回答标记为完成并刷新元数据。
error可以公开的错误消息结束加载状态并显示重试入口。

完整异常应记录在服务端，但不能把堆栈、API Key 或供应商原始响应传给浏览器。

第 5 步：在 React 中读取 SSE​

聊天接口使用 POST，所以要用 fetch() 读取响应流，而不是使用只支持 GET 的浏览器 EventSource API。

const response = await fetch(`/api/conversations/${conversationId}/messages/stream`, {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({message}),
});
if (!response.ok || !response.body) throw new Error('流式请求失败');

const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';

while (true) {
  const {value, done} = await reader.read();
  buffer += decoder.decode(value, {stream: !done}).replace(/\r\n/g, '\n');
  const blocks = buffer.split('\n\n');
  buffer = blocks.pop() ?? '';

  for (const block of blocks) {
    const event = block.match(/^event:\s*(.+)$/m)?.[1] ?? 'message';
    const raw = block.match(/^data:\s*(.+)$/m)?.[1];
    if (!raw) continue;
    const data = JSON.parse(raw);

    if (event === 'delta') appendAssistantText(data.text);
    if (event === 'chart') showChartTools(data.tools);
    if (event === 'done') finishAssistantMessage();
    if (event === 'error') throw new Error(data.message);
  }
  if (done) break;
}

完整 Demo 已经处理了更健壮的流解析、乐观消息、Markdown 渲染、加载状态和错误恢复。生产实现可以直接参考 Python Demo 的 App.tsx 或 TypeScript Demo 的 App.tsx。

第 6 步：补齐会话生命周期​

围绕 ChatSession 实现这些接口：

Method路径SDK 操作
GET/api/conversations按已认证的 external_user_id 列出对话。
POST/api/conversations创建 ChatSession，再保存本地元数据。
GET/api/conversations/:id校验归属，再调用 get_items / getItems。
PATCH/api/conversations/:id校验归属后修改本地界面标题。
DELETE/api/conversations/:id先调用 clear_session / clearSession，再删除本地元数据。
POST/api/conversations/:id/forkFork 托管上下文，并为新 ID 创建元数据。
POST/api/conversations/:id/messages/stream运行下一轮并返回 SSE。

删除顺序很重要：先删除托管对话，再删除本地映射。否则供应商调用失败时，可能留下一个无法从产品中发现的托管对话。生命周期方法和两类 ID 的边界请见 ChatSession 管理。

第 7 步：通过工具连接自己的数据库​

如果 Agent 需要读取订阅状态、预约或已保存资料，应提供一个范围很小的后端工具。工具只接收已经校验的参数；用户身份从可信的服务端上下文获取，不能让模型自己选择用户 ID。

工具输出应尽量少、结构清晰且不包含秘密。扣费、预约、发送和删除等操作必须先取得用户确认或人工批准，再产生副作用。

目前有一项接入边界需要提前规划：托管 Iztro 工具可以流式运行；开发者自定义工具的循环应使用非流式调用。请阅读非流式调用和工具。你仍可以由自己的后端发送状态事件，但不要假设 SDK 能在同一次运行中把所有自定义工具循环与托管流式输出组合起来。

第 8 步：验证完整流程​

准备两个用户，按以下顺序测试：

  1. 用户 A 新建会话并发送消息。
  2. 刷新浏览器，确认同一会话能带着上下文继续。
  3. 用户 B 无法读取、发消息、改名、Fork 或删除用户 A 的会话。
  4. 只有真正执行 Iztro 命盘工具时，页面才会收到 chart 事件。
  5. 删除会话后，托管上下文和本地元数据都被清除。
  6. 浏览器网络响应和构建后的 JavaScript 中没有 API Key。

修改 Demo 后运行自带检查：

pytest examples/fullstack-demo/backend/tests -q
cd examples/fullstack-demo/frontend
npm run build

上线前检查​

  • 把 ZIWEI_API_KEY 放在密钥管理器或后端环境变量中；一旦泄露立即轮换。
  • 从已验证的登录态得到 external_user_id，每个会话接口都校验归属。
  • 使用 HTTPS、明确的 CORS 白名单、请求大小限制、限流和防滥用措施。
  • 对 SSE 关闭反向代理缓冲，并把空闲超时设置得足够长。
  • 校验所有工具参数；产生副作用时使用事务和幂等键。
  • 同时定义托管对话和本地记录的保留期限与账号删除流程。
  • 对敏感出生资料保存用户同意记录和访问规则。
  • 增加请求 ID 与服务端错误日志，但不要记录密钥或不必要的个人数据。

至此，系统边界就很清楚了：React 负责交互，你的后端负责信任与业务操作，ChatSession 负责对话上下文，你的数据库负责长期有效的产品事实。