从"你问我答"到"实时对话"：交互模型如何重塑人机协作

用户输入 → 模型推理 → 返回完整文本 → 等待下一轮输入

"""
实时流式协作示例：模拟交互模型的"中途干预"能力
依赖：pip install openai fastapi uvicorn websockets
运行：python interaction_sim.py
然后打开浏览器访问 ws://localhost:8000/ws 进行交互
"""

import asyncio
import json
from fastapi import FastAPI, WebSocket
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()  # 需设置 OPENAI_API_KEY 环境变量

# 共享的持续上下文——交互模型的核心特征
conversation_context: list[dict] = []


@app.websocket("/ws")
async def interaction_ws(ws: WebSocket):
    await ws.accept()
    await ws.send_json({"type": "system", "content": "连接建立，你可以随时输入。输入 /redirect <新方向> 可中途改变生成方向。"})

    # 用户输入队列：支持持续流入
    user_queue: asyncio.Queue = asyncio.Queue()
    redirect_flag: asyncio.Event = asyncio.Event()
    redirect_msg: str = ""

    async def receive_user_input():
        """持续接收用户输入，区分正常消息和干预指令"""
        try:
            while True:
                data = await ws.receive_text()
                msg = json.loads(data)
                text = msg.get("content", "")
                if text.startswith("/redirect"):
                    # 中途干预：设置重定向标志
                    redirect_msg = text.replace("/redirect", "").strip()
                    redirect_flag.set()
                else:
                    await user_queue.put(text)
        except Exception:
            pass

    async def stream_with_intervention():
        """流式生成，支持中途被用户干预打断并转向"""
        while True:
            # 等待用户的新输入
            user_text = await user_queue.get()
            if user_text is None:
                break

            conversation_context.append({"role": "user", "content": user_text})

            # 开始流式生成
            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=conversation_context,
                stream=True,
            )

            accumulated = ""
            redirect_flag.clear()

            async for chunk in stream:
                # 检查用户是否发出中途干预
                if redirect_flag.is_set():
                    # 把干预信息注入上下文，重新生成
                    await ws.send_json({
                        "type": "interrupted",
                        "content": f"收到方向修正：{redirect_msg}，正在转向..."
                    })
                    conversation_context.append({
                        "role": "assistant", "content": accumulated + f"\n[用户中途修正：{redirect_msg}]"
                    })
                    conversation_context.append({
                        "role": "user", "content": redirect_msg
                    })
                    # 用修正后的上下文重新开始生成
                    new_stream = await client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=conversation_context,
                        stream=True,
                    )
                    accumulated = ""
                    async for new_chunk in new_stream:
                        delta = new_chunk.choices[0].delta.content or ""
                        accumulated += delta
                        await ws.send_json({"type": "token", "content": delta})
                    break

                delta = chunk.choices[0].delta.content or ""
                accumulated += delta
                await ws.send_json({"type": "token", "content": delta})

            # 保存完整输出到持续上下文
            conversation_context.append({"role": "assistant", "content": accumulated})
            await ws.send_json({"type": "done", "content": accumulated})

    # 并行运行：输入接收与生成输出同时进行
    recv_task = asyncio.create_task(receive_user_input())
    gen_task = asyncio.create_task(stream_with_intervention())

    try:
        await asyncio.gather(recv_task, gen_task)
    except Exception:
        recv_task.cancel()
        gen_task.cancel()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)