实验 02 — 你的第一个流式聊天¶

目标：跟随一个提示，从 Python 堆栈的客户端 HTTP 到 FastAPI 再到 Copilot SDK，直到模型，理解为什么模型发现是运行时数据而非硬编码的列表。

用时： ~20分钟

前置条件： 实验 01 完成，FastAPI 服务器正在端口 5070 上运行。

步骤 1 — 启动或确认服务器¶

从仓库根目录：

cd src/AgentOrchestrator-python
uv run uvicorn app.main:app --port 5070

打开 http://localhost:5070 如果您想稍后观看 UI。相同的流程同时提供 API 和静态 UI。

确认 API 是健康的：

curl http://localhost:5070/api/chat/health

预期结果：

{"status":"healthy","service":"CopilotChat","availableModels":["claude-haiku-4.5","gpt-4.1","gpt-5","claude-sonnet-4.5","claude-opus-4.5","gemini-2.5-pro"]}

那 fallback 列表是故意很小的。真正的模型选择器会询问已登录的 Copilot 账户它能使用什么。

步骤 2 — 观察传输格式¶

发送提示并观察原始的 Server-Sent Events：

curl -sN -X POST http://localhost:5070/api/chat/stream \
    -H 'Content-Type: application/json' \
    -d '{"prompt":"Reply with exactly: streaming works","model":"claude-haiku-4.5"}'

捕获输出：

data: {"content": "streaming works"}

data: [DONE]

三件事要注意：每一帧是data:加上JSON，后面跟着一个
空行；较长的回答以多个帧的形式到来，因为片段可以在任何地方分割；并且流以data: [DONE]结束。

⚠️ 产品名的请求模型是 prompt, model, 和可选的 systemMessage。一个未识别的键，如 message，将被默默地忽略——你将收到一个通用的问候而不是验证错误。

步骤 3 — 找到服务器端¶

打开app/routers/chat.py，并定位到stream_chat。

stream_chat 返回一个 StreamingResponse，其媒体类型为 text/event-stream。在 event_stream 内，每个 SDK 块都被序列化为一个 SSE 帧：

yield f"data: {json.dumps({'content': chunk})}\n\n"

该路由接着写 data: [DONE] 终止标记。

⚠️ 空行不是装饰。它是事件分隔符。删掉它，许多 SSE 客户端会继续缓冲，因为它们从未看到一个完整的事件。

现在看看这个 except 块。错误被写入流中作为data: {"error": "..."}<code>data: {"error": "..."}</code>

步骤 4 — 找到 SDK 集成¶

打开 app/services/copilot_chat.py.

CopilotChatService.chat_stream 创建了一个启用流式传输的会话：

session = await self._client.create_session(
    model=model,
    streaming=True,
    system_message=(
        {"mode": "append", "content": system_message} if system_message else None
    ),
)

然后它订阅事件：

def on_event(evt: SessionEvent) -> None:
    # Unlike .NET, every event arrives as one SessionEvent
    # carrying a `type` enum and a `data` payload, so this
    # dispatches on `evt.type` rather than on subclasses.
    if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
        queue.put_nowait(evt.data.delta_content or "")
    elif evt.type is SessionEventType.ASSISTANT_MESSAGE:
        logger.info(
            "Assistant response complete: %d chars",
            len(evt.data.content or ""),
        )
    elif evt.type is SessionEventType.SESSION_IDLE:
        if not done.done():
            done.set_result(None)
    elif evt.type is SessionEventType.SESSION_ERROR:
        logger.error("Session error: %s", evt.data.message)
        if not done.done():
            done.set_exception(RuntimeError(evt.data.message))

The queue是SDK的回调风格和FastAPI的异步响应生成器之间的桥梁。回调将块推入asyncio.Queue；路由等待队列并生成SSE帧。

步骤 5 — 理解 Python 事件¶

这是 Python 和 .NET 之间最重要的一个区别。

.NET 示例会对事件子类进行模式匹配，例如 AssistantMessageDeltaEvent 和 SessionIdleEvent. Python 给你一个 SessionEvent 数据类，其中包含如下的字段：

type
data
id
timestamp

该 type字段是一个SessionEventType枚举，因此Python代码分支如下：

if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
    ...
elif evt.type is SessionEventType.SESSION_ERROR:
    ...

事件也是只推送回调。session.on(handler) 返回一个取消订阅的可调用函数；没有直接遍历的异步迭代器。

该助手在 sdk_labs/_common.py 使用这种模式等待直到 SESSION_IDLE 或抛出异常 SESSION_ERROR.

第 6 步 — 询问 API 您有哪些模型¶

curl -s http://localhost:5070/api/chat/models | jq -r '.[].id'

/api/chat/models 返回一个 JSON 列表，其对象的结构如下：

{"id":"...","name":"...","description":"..."}

路由首先调用 CopilotChatService.list_models()，调用 await client.list_models()那返回 ModelInfo ModelInfo对象包含.id和.name字段。 .id 以及 .name 字段。

⚠️ list_models() 已知有上游 Bug：它能引发 ValueError: Missing required field 'multiplier' in ModelBilling (github/copilot-sdk#1302)。路由广泛捕获并退回到静态目录，因此 UI 仍然有选择。

第 7 步 — 切换模型并比较¶

从您的活列表中选择一个模型 ID：

MODEL=$(curl -s http://localhost:5070/api/chat/models | jq -r '.[0].id')
echo "Using $MODEL"

curl -sN -X POST http://localhost:5070/api/chat/stream \
  -H 'Content-Type: application/json' \
  -d "{\"prompt\":\"In one sentence, what is customer churn?\",\"model\":\"$MODEL\"}"

使用不同的 id 重复，并比较延迟、样式和语气。

The 智能体选择器在 sdk_labs/model_picker.py 使用了相同的概念：优先使用 claude-haiku-4.5，但会退回到你实际可以使用的具体模型。你可以使用 --model <id> 来覆盖示例。

步骤 8 — 使用系统消息塑造响应¶

The API 接受可选 systemMessage服务将其发送在追加模式，以便将其补充到会话的内置指令中，而不是替换它们。

curl -sN -X POST http://localhost:5070/api/chat/stream \
  -H 'Content-Type: application/json' \
  -d '{
    "prompt":"Which segment has the lowest retention?",
    "model":"claude-haiku-4.5",
    "systemMessage":"You are a retail analytics assistant. Context: 4 segments — High Value (92% retention), Regular (78%), At Risk (45%), New (65%). Answer in one sentence."
  }'

预期行为：一个以数据为依据的答案命名面临风险，占比45%。如果没有这种上下文，模型无法直接访问你的种子数据。实验03用一个模型只能在需要零售事实时调用的工具替换静态上下文。

步骤 9 — 了解基本会话调用¶

The Python SDK 的对象支持 async with，因此示例在客户端和会话中都确定性地清理。

对于非流式传输的一次性提示，SDK 也提供了：

await session.send_and_wait(prompt, timeout=60.0)

The FastAPI服务使用send(prompt)，因为它会处理每个delta作为它们到达时。

第 10 步 —— 尝试 UI 路径¶

回到浏览器中的 http://localhost:5070，选择一个模型，问 "列出三个零售关键绩效指标。一行一条。"，然后观察消息以块的形式渲染。如果流失败，检查 data: {"error": "..."} 帧。

完成的交换看起来是这样的——与上述通过 curl 读取的 SSE 帧相同，由 app.js 解析并渲染为 Markdown：

该聊天界面在询问用户'列出三个零售关键绩效指标(KPI)，每项一条。'后，助手回复了一个带编号的列表：转化率、平均订单价值（AOV）和客户保留率，每项都附带了一条简短的定义。

💡 在流中的助手气泡显示一个动画的输入指示器；一旦到达 [DONE] 标记符，它就会被渲染的 Markdown 替换。

⚠️ 静态 UI 发布 { prompt, model } 从 app/static/app.js，匹配 ChatRequest。一个较早的版本发送了 message；由于 Pydantic 忽略未知键而不是拒绝它们，浏览器将回复流式传输到一个空提示，并且没有大声失败。现在 tests/test_chat_contract.py 断言 app.js 发布的字段正是 API 读取的字段。

✅ 检查点¶

您可以现在解释：

[x] SSE 传输格式以及空行为何重要
[x] 为什么错误是流式传输而不是通过 HTTP 状态码返回
[x] 如 FastAPI 的 StreamingResponse 包裹了 SDK 的流
[x] 为什么 Python 通过 evt.type 分支而不是事件子类
[x] 模型是在运行时被发现的
[x] 如系统消息使助手在零售领域以数据为依据

💡 拓展练习¶

打开 app/services/copilot_chat.py，并在if链之前记录每种事件类型。发送一个简短的提示，并将事件序列与助手示例进行比较：

uv run python -m sdk_labs events

那更深的事件生命周期示例是sdk_labs/events_sample.py。