streaming-llm-responses

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Streaming LLM Responses

流式LLM响应

Build responsive, real-time chat interfaces with streaming feedback.

构建带流式反馈的响应式实时聊天界面。

Quick Start

快速开始

typescript

import { useChatKit } from "@openai/chatkit-react";

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onResponseStart: () => setIsResponding(true),
  onResponseEnd: () => setIsResponding(false),

  onEffect: ({ name, data }) => {
    if (name === "update_status") updateUI(data);
  },
});

typescript

import { useChatKit } from "@openai/chatkit-react";

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onResponseStart: () => setIsResponding(true),
  onResponseEnd: () => setIsResponding(false),

  onEffect: ({ name, data }) => {
    if (name === "update_status") updateUI(data);
  },
});

Response Lifecycle

响应生命周期

User sends message
    ↓
onResponseStart() fires
    ↓
[Streaming: tokens arrive, ProgressUpdateEvents shown]
    ↓
onResponseEnd() fires
    ↓
UI unlocks, ready for next interaction

用户发送消息
    ↓
onResponseStart() 触发
    ↓
[流式传输：令牌到达，显示ProgressUpdateEvents]
    ↓
onResponseEnd() 触发
    ↓
UI解锁，准备下一次交互

Core Patterns

核心模式

1. Response Lifecycle Handlers

1. 响应生命周期处理程序

Lock UI during AI response to prevent race conditions:

typescript

function ChatWithLifecycle() {
  const [isResponding, setIsResponding] = useState(false);
  const lockInteraction = useAppStore((s) => s.lockInteraction);
  const unlockInteraction = useAppStore((s) => s.unlockInteraction);

  const chatkit = useChatKit({
    api: { url: API_URL, domainKey: DOMAIN_KEY },

    onResponseStart: () => {
      setIsResponding(true);
      lockInteraction(); // Disable map/canvas/form interactions
    },

    onResponseEnd: () => {
      setIsResponding(false);
      unlockInteraction();
    },

    onError: ({ error }) => {
      console.error("ChatKit error:", error);
      setIsResponding(false);
      unlockInteraction();
    },
  });

  return (
    <div>
      {isResponding && <LoadingOverlay />}
      <ChatKit control={chatkit.control} />
    </div>
  );
}

在AI响应期间锁定UI以防止竞态条件：

typescript

function ChatWithLifecycle() {
  const [isResponding, setIsResponding] = useState(false);
  const lockInteraction = useAppStore((s) => s.lockInteraction);
  const unlockInteraction = useAppStore((s) => s.unlockInteraction);

  const chatkit = useChatKit({
    api: { url: API_URL, domainKey: DOMAIN_KEY },

    onResponseStart: () => {
      setIsResponding(true);
      lockInteraction(); // 禁用地图/画布/表单交互
    },

    onResponseEnd: () => {
      setIsResponding(false);
      unlockInteraction();
    },

    onError: ({ error }) => {
      console.error("ChatKit error:", error);
      setIsResponding(false);
      unlockInteraction();
    },
  });

  return (
    <div>
      {isResponding && <LoadingOverlay />}
      <ChatKit control={chatkit.control} />
    </div>
  );
}

2. Client Effects (Fire-and-Forget)

2. 客户端效果（即发即弃）

Server sends effects to update client UI without expecting a response:

Backend - Streaming Effects:

python

from chatkit.types import ClientEffectEvent

async def respond(self, thread, item, context):
    # ... agent processing ...

    # Fire client effect to update UI
    yield ClientEffectEvent(
        name="update_status",
        data={
            "state": {"energy": 80, "happiness": 90},
            "flash": "Status updated!"
        }
    )

    # Another effect
    yield ClientEffectEvent(
        name="show_notification",
        data={"message": "Task completed!"}
    )

Frontend - Handling Effects:

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onEffect: ({ name, data }) => {
    switch (name) {
      case "update_status":
        applyStatusUpdate(data.state);
        if (data.flash) setFlashMessage(data.flash);
        break;

      case "add_marker":
        addMapMarker(data);
        break;

      case "select_mode":
        setSelectionMode(data.mode);
        break;
    }
  },
});

服务器发送效果以更新客户端UI，无需等待响应：

后端 - 流式效果：

python

from chatkit.types import ClientEffectEvent

async def respond(self, thread, item, context):
    # ... agent处理 ...

    # 触发客户端效果以更新UI
    yield ClientEffectEvent(
        name="update_status",
        data={
            "state": {"energy": 80, "happiness": 90},
            "flash": "Status updated!"
        }
    )

    # 另一个效果
    yield ClientEffectEvent(
        name="show_notification",
        data={"message": "Task completed!"}
    )

前端 - 处理效果：

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onEffect: ({ name, data }) => {
    switch (name) {
      case "update_status":
        applyStatusUpdate(data.state);
        if (data.flash) setFlashMessage(data.flash);
        break;

      case "add_marker":
        addMapMarker(data);
        break;

      case "select_mode":
        setSelectionMode(data.mode);
        break;
    }
  },
});

3. Progress Updates

3. 进度更新

Show "Searching...", "Loading...", "Analyzing..." during long operations:

python

from chatkit.types import ProgressUpdateEvent

@function_tool
async def search_articles(ctx: AgentContext, query: str) -> str:
    """Search for articles matching the query."""

    yield ProgressUpdateEvent(message="Searching articles...")

    results = await article_store.search(query)

    yield ProgressUpdateEvent(message=f"Found {len(results)} articles...")

    for i, article in enumerate(results):
        if i % 5 == 0:
            yield ProgressUpdateEvent(
                message=f"Processing article {i+1}/{len(results)}..."
            )

    return format_results(results)

在长操作期间显示“搜索中...”, “加载中...”, “分析中...”：

python

from chatkit.types import ProgressUpdateEvent

@function_tool
async def search_articles(ctx: AgentContext, query: str) -> str:
    """Search for articles matching the query."""

    yield ProgressUpdateEvent(message="Searching articles...")

    results = await article_store.search(query)

    yield ProgressUpdateEvent(message=f"Found {len(results)} articles...")

    for i, article in enumerate(results):
        if i % 5 == 0:
            yield ProgressUpdateEvent(
                message=f"Processing article {i+1}/{len(results)}..."
            )

    return format_results(results)

4. Thread Lifecycle Events

4. 线程生命周期事件

Track thread changes for persistence and UI updates:

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onThreadChange: ({ threadId }) => {
    setThreadId(threadId);
    if (threadId) localStorage.setItem("lastThreadId", threadId);
    clearSelections();
  },

  onThreadLoadStart: ({ threadId }) => {
    setIsLoadingThread(true);
  },

  onThreadLoadEnd: ({ threadId }) => {
    setIsLoadingThread(false);
  },
});

跟踪线程变化以实现持久化和UI更新：

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onThreadChange: ({ threadId }) => {
    setThreadId(threadId);
    if (threadId) localStorage.setItem("lastThreadId", threadId);
    clearSelections();
  },

  onThreadLoadStart: ({ threadId }) => {
    setIsLoadingThread(true);
  },

  onThreadLoadEnd: ({ threadId }) => {
    setIsLoadingThread(false);
  },
});

5. Client Tools (State Query)

5. 客户端工具（状态查询）

AI needs to read client-side state to make decisions:

Backend - Defining Client Tool:

python

@function_tool(name_override="get_selected_items")
async def get_selected_items(ctx: AgentContext) -> dict:
    """Get the items currently selected on the canvas.

    This is a CLIENT TOOL - executed in browser, result comes back.
    """
    yield ProgressUpdateEvent(message="Reading selection...")
    pass  # Actual execution happens on client

Frontend - Handling Client Tools:

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onClientTool: ({ name, params }) => {
    switch (name) {
      case "get_selected_items":
        return { itemIds: selectedItemIds };

      case "get_current_viewport":
        return {
          center: mapRef.current.getCenter(),
          zoom: mapRef.current.getZoom(),
        };

      case "get_form_data":
        return { values: formRef.current.getValues() };

      default:
        throw new Error(`Unknown client tool: ${name}`);
    }
  },
});

AI需要读取客户端状态以做出决策：

后端 - 定义客户端工具：

python

@function_tool(name_override="get_selected_items")
async def get_selected_items(ctx: AgentContext) -> dict:
    """Get the items currently selected on the canvas.

    This is a CLIENT TOOL - executed in browser, result comes back.
    """
    yield ProgressUpdateEvent(message="Reading selection...")
    pass  # 实际执行在客户端进行

前端 - 处理客户端工具：

typescript

const chatkit = useChatKit({
  api: { url: API_URL, domainKey: DOMAIN_KEY },

  onClientTool: ({ name, params }) => {
    switch (name) {
      case "get_selected_items":
        return { itemIds: selectedItemIds };

      case "get_current_viewport":
        return {
          center: mapRef.current.getCenter(),
          zoom: mapRef.current.getZoom(),
        };

      case "get_form_data":
        return { values: formRef.current.getValues() };

      default:
        throw new Error(`Unknown client tool: ${name}`);
    }
  },
});

Client Effects vs Client Tools

客户端效果 vs 客户端工具

Type	Direction	Response Required	Use Case
Client Effect	Server → Client	No (fire-and-forget)	Update UI, show notifications
Client Tool	Server → Client → Server	Yes (return value)	Get client state for AI decision

类型	方向	是否需要响应	适用场景
客户端效果	服务器 → 客户端	否（即发即弃）	更新UI、显示通知
客户端工具	服务器 → 客户端 → 服务器	是（需返回值）	获取客户端状态供AI决策

Common Patterns by Use Case

按场景划分的常见模式

Interactive Map/Canvas

交互式地图/画布

typescript

onResponseStart: () => lockCanvas(),
onResponseEnd: () => unlockCanvas(),
onEffect: ({ name, data }) => {
  if (name === "add_marker") addMarker(data);
  if (name === "pan_to") panTo(data.location);
},
onClientTool: ({ name }) => {
  if (name === "get_selection") return getSelectedItems();
},

typescript

onResponseStart: () => lockCanvas(),
onResponseEnd: () => unlockCanvas(),
onEffect: ({ name, data }) => {
  if (name === "add_marker") addMarker(data);
  if (name === "pan_to") panTo(data.location);
},
onClientTool: ({ name }) => {
  if (name === "get_selection") return getSelectedItems();
},

Form-Based UI

基于表单的UI

typescript

onResponseStart: () => setFormDisabled(true),
onResponseEnd: () => setFormDisabled(false),
onClientTool: ({ name }) => {
  if (name === "get_form_values") return form.getValues();
},

typescript

onResponseStart: () => setFormDisabled(true),
onResponseEnd: () => setFormDisabled(false),
onClientTool: ({ name }) => {
  if (name === "get_form_values") return form.getValues();
},

Game/Simulation

游戏/模拟

typescript

onResponseStart: () => pauseSimulation(),
onResponseEnd: () => resumeSimulation(),
onEffect: ({ name, data }) => {
  if (name === "update_entity") updateEntity(data);
  if (name === "show_notification") showToast(data.message);
},

typescript

onResponseStart: () => pauseSimulation(),
onResponseEnd: () => resumeSimulation(),
onEffect: ({ name, data }) => {
  if (name === "update_entity") updateEntity(data);
  if (name === "show_notification") showToast(data.message);
},

Thread Title Generation

线程标题生成

Dynamically update thread title based on conversation:

python

class TitleAgent:
    async def generate_title(self, first_message: str) -> str:
        result = await Runner.run(
            Agent(
                name="TitleGenerator",
                instructions="Generate a 3-5 word title.",
                model="gpt-4o-mini",  # Fast model
            ),
            input=f"First message: {first_message}",
        )
        return result.final_output

根据对话动态更新线程标题：

python

class TitleAgent:
    async def generate_title(self, first_message: str) -> str:
        result = await Runner.run(
            Agent(
                name="TitleGenerator",
                instructions="Generate a 3-5 word title.",
                model="gpt-4o-mini",  # 快速模型
            ),
            input=f"First message: {first_message}",
        )
        return result.final_output

In ChatKitServer

在ChatKitServer中

async def respond(self, thread, item, context): if not thread.title and item: title = await self.title_agent.generate_title(item.content) thread.title = title await self.store.save_thread(thread, context)

---

---

Anti-Patterns

反模式

Not locking UI during response - Leads to race conditions
Blocking in effects - Effects should be fire-and-forget
Heavy computation in onEffect - Use requestAnimationFrame for DOM updates
Missing error handling - Always handle onError to unlock UI
Not persisting thread state - Use onThreadChange to save context

响应期间未锁定UI - 会导致竞态条件
在效果中阻塞 - 效果应是即发即弃的
在onEffect中执行大量计算 - DOM更新使用requestAnimationFrame
缺少错误处理 - 务必处理onError以解锁UI
未持久化线程状态 - 使用onThreadChange保存上下文

Verification

验证

Run:

python3 scripts/verify.py

Expected:

✓ streaming-llm-responses skill ready

运行：

python3 scripts/verify.py

预期结果：

✓ streaming-llm-responses skill ready

If Verification Fails

若验证失败

Check: references/ folder has streaming-patterns.md
Stop and report if still failing

检查：references/ 文件夹是否包含streaming-patterns.md
若仍失败，请停止并上报

References

参考资料

references/streaming-patterns.md - Complete streaming configuration

references/streaming-patterns.md - 完整的流式配置文档