流式输入、增量理解与实时响应

from openai import OpenAI
import json

client = OpenAI(api_key="your-api-key")

def stream_llm_response_streaming(prompt, model="gpt-4-turbo"):
    """
    使用 SSE 流式输出 LLM 回复
    
    Args:
        prompt: 用户输入
        model: LLM 模型
    
    Yields:
        token: 逐个返回生成的 Token
    """
    
    # 创建流式请求
    stream = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt}
        ],
        stream=True,  # 启用流式输出
        temperature=0.7
    )
    
    # 逐个处理返回的 chunk
    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            token = chunk.choices[0].delta.content
            yield token  # 逐 Token 返回


def display_streaming_response(prompt):
    """
    显示流式回复（模拟终端逐字显示）
    
    Args:
        prompt: 用户输入
    """
    import sys
    import time
    
    print(f"用户：{prompt}\n")
    print("助手：", end="", flush=True)
    
    # 记录开始时间
    start_time = time.time()
    token_count = 0
    first_token_time = None
    
    # 流式获取并显示 Token
    for token in stream_llm_response_streaming(prompt):
        # 记录首字时间
        if first_token_time is None:
            first_token_time = time.time()
            ttft = (first_token_time - start_time) * 1000  # 毫秒
            print(f"\n[首字延迟：{ttft:.0f}ms]\n", end="", flush=True)
        
        # 逐字显示
        print(token, end="", flush=True)
        token_count += 1
        
        # 模拟打字效果（可选）
        # time.sleep(0.02)
    
    # 计算总时间
    total_time = (time.time() - start_time) * 1000  # 毫秒
    tokens_per_second = token_count / (total_time / 1000)
    
    print(f"\n\n[生成完成：{token_count} tokens, {total_time:.0f}ms, {tokens_per_second:.1f} tokens/s]")
    
    return {
        "ttft_ms": ttft,
        "total_time_ms": total_time,
        "token_count": token_count,
        "tokens_per_second": tokens_per_second
    }


# WebSocket 流式输出示例（前端实时显示）
def websocket_streaming_example():
    """
    WebSocket 流式输出示例
    
    后端通过 WebSocket 向前端推送 Token
    """
    
    # 后端代码（FastAPI + WebSocket）
    backend_code = '''
from fastapi import FastAPI, WebSocket
from openai import OpenAI

app = FastAPI()
client = OpenAI(api_key="your-api-key")

@app.websocket("/ws/stream")
async def websocket_stream(websocket: WebSocket):
    await websocket.accept()
    
    # 接收用户输入
    user_input = await websocket.receive_text()
    
    # 创建流式请求
    stream = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": user_input}],
        stream=True
    )
    
    # 逐 Token 推送给前端
    for chunk in stream:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            await websocket.send_json({"type": "token", "content": token})
    
    # 发送完成信号
    await websocket.send_json({"type": "done"})
'''
    
    # 前端代码（JavaScript）
    frontend_code = '''
// 前端 JavaScript 代码
const ws = new WebSocket("ws://localhost:8000/ws/stream");

let responseText = "";

ws.onopen = () => {
    // 发送用户输入
    ws.send("请解释量子纠缠");
};

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    
    if (data.type === "token") {
        // 逐 Token 追加显示
        responseText += data.content;
        document.getElementById("response").textContent = responseText;
        
        // 自动滚动到底部
        window.scrollTo(0, document.body.scrollHeight);
    } else if (data.type === "done") {
        console.log("生成完成");
        ws.close();
    }
};
'''
    
    print("WebSocket 流式输出示例:")
    print("\n后端代码:")
    print(backend_code)
    print("\n" + "="*60 + "\n")
    print("前端代码:")
    print(frontend_code)


if __name__ == "__main__":
    # 示例 1：终端流式显示
    print("="*60)
    print("示例 1：终端流式显示")
    print("="*60 + "\n")
    
    prompt = "请用简洁的语言解释什么是量子纠缠"
    stats = display_streaming_response(prompt)
    
    print("\n" + "="*60 + "\n")
    
    # 示例 2：WebSocket 流式
    print("示例 2：WebSocket 流式输出")
    print("="*60)
    websocket_streaming_example()

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import torch.nn.functional as F

class StreamingIntentRecognizer:
    """
    流式意图识别器
    
    基于部分输入实时预测意图，并随新输入动态更新
    """
    
    def __init__(self, model_name="bert-base-uncased", intent_labels=None):
        """
        初始化
        
        Args:
            model_name: 预训练模型名称
            intent_labels: 意图标签列表
        """
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, 
            num_labels=len(intent_labels) if intent_labels else 10
        )
        self.intent_labels = intent_labels or ["query", "command", "question", "statement", "other"]
        
        # 设置模型为评估模式
        self.model.eval()
    
    def recognize_partial(self, partial_text, confidence_threshold=0.6):
        """
        基于部分文本识别意图
        
        Args:
            partial_text: 部分输入文本
            confidence_threshold: 置信度阈值
        
        Returns:
            result: {
                "intent": 预测意图,
                "confidence": 置信度,
                "is_stable": 是否稳定（置信度超过阈值）,
                "all_probabilities": 所有意图概率分布
            }
        """
        # Tokenize 输入
        inputs = self.tokenizer(
            partial_text,
            return_tensors="pt",
            truncation=True,
            max_length=128,
            padding=True
        )
        
        # 模型推理
        with torch.no_grad():
            outputs = self.model(**inputs)
            probabilities = F.softmax(outputs.logits, dim=-1)[0]
        
        # 获取最高概率意图
        max_prob, predicted_idx = torch.max(probabilities, dim=0)
        predicted_intent = self.intent_labels[predicted_idx.item()]
        confidence = max_prob.item()
        
        # 判断是否稳定
        is_stable = confidence >= confidence_threshold
        
        return {
            "intent": predicted_intent,
            "confidence": confidence,
            "is_stable": is_stable,
            "all_probabilities": {
                label: prob.item() 
                for label, prob in zip(self.intent_labels, probabilities)
            }
        }
    
    def streaming_recognition(self, text_chunks):
        """
        流式识别：处理一系列文本块
        
        Args:
            text_chunks: 文本块列表（按时间顺序）
        
        Yields:
            result: 每次的识别结果
        """
        accumulated_text = ""
        previous_intent = None
        
        for i, chunk in enumerate(text_chunks):
            # 累积文本
            accumulated_text += chunk
            
            # 识别当前意图
            result = self.recognize_partial(accumulated_text)
            result["chunk_index"] = i
            result["accumulated_text"] = accumulated_text
            
            # 检测意图变化
            if previous_intent is not None and result["intent"] != previous_intent:
                result["intent_changed"] = True
                result["previous_intent"] = previous_intent
            else:
                result["intent_changed"] = False
            
            previous_intent = result["intent"]
            
            yield result


# 使用示例
if __name__ == "__main__":
    # 初始化识别器
    recognizer = StreamingIntentRecognizer(
        intent_labels=[
            "book_flight",      # 预订机票
            "check_weather",    # 查询天气
            "set_reminder",     # 设置提醒
            "play_music",       # 播放音乐
            "question",         # 一般问题
            "other"             # 其他
        ]
    )
    
    # 模拟流式输入（用户边说话边识别）
    text_chunks = [
        "帮我",           # 块 1
        "帮我订一张",     # 块 2
        "帮我订一张明天",  # 块 3
        "帮我订一张明天去",  # 块 4
        "帮我订一张明天去北京的",  # 块 5
        "帮我订一张明天去北京的机票"  # 块 6（完整）
    ]
    
    print("流式意图识别示例:")
    print("="*70 + "\n")
    
    # 流式处理
    for result in recognizer.streaming_recognition(text_chunks):
        status = "✓ 稳定" if result["is_stable"] else "○ 不确定"
        change = f" (变化：{result['previous_intent']} → {result['intent']})" if result.get("intent_changed") else ""
        
        print(f"块 {result['chunk_index'] + 1}: \"{result['accumulated_text']}\"")
        print(f"  意图：{result['intent']} (置信度：{result['confidence']:.2f}) [{status}]{change}")
        print()
    
    print("="*70)
    print("\n关键观察:")
    print("1. 早期块（1-2）置信度低，意图不稳定")
    print("2. 中期块（3-4）置信度提升，意图趋于稳定")
    print("3. 后期块（5-6）置信度高，意图稳定为 'book_flight'")
    print("4. 系统可在置信度超过阈值时提前开始准备响应")

from openai import OpenAI
import json

client = OpenAI(api_key="your-api-key")

def streaming_chain_of_thought(question, model="gpt-4-turbo"):
    """
    流式思维链推理
    
    逐步生成推理步骤，每步立即返回
    支持用户中途查看推理过程并干预
    
    Args:
        question: 问题
        model: LLM 模型
    
    Yields:
        step: 推理步骤（包含步骤编号、内容、中间结论）
    """
    
    # 构建提示词，要求逐步推理
    prompt = f"""
请逐步推理以下问题，每一步都清晰标注步骤编号。

问题：{question}

推理步骤：
"""
    
    # 创建流式请求
    stream = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt}
        ],
        stream=True,
        temperature=0.2  # 低温度保证推理稳定
    )
    
    accumulated_text = ""
    current_step = 1
    step_content = ""
    
    # 逐 Token 处理
    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            token = chunk.choices[0].delta.content
            accumulated_text += token
            step_content += token
            
            # 检测步骤标记（如"步骤 1:"、"Step 2:"等）
            step_markers = ["步骤", "Step", "step", "推理", "首先", "然后", "最后"]
            
            # 检测步骤结束（换行或下一个步骤标记）
            if "\n" in token or any(marker in token for marker in step_markers[4:]):
                if step_content.strip():
                    # 提取中间结论（最后一句）
                    sentences = step_content.strip().split("。")
                    intermediate_conclusion = sentences[-1] if sentences else step_content.strip()
                    
                    yield {
                        "step_number": current_step,
                        "content": step_content.strip(),
                        "intermediate_conclusion": intermediate_conclusion,
                        "is_final": False
                    }
                    
                    current_step += 1
                    step_content = ""
    
    # 最后一步
    if step_content.strip():
        sentences = step_content.strip().split("。")
        final_conclusion = sentences[-1] if sentences else step_content.strip()
        
        yield {
            "step_number": current_step,
            "content": step_content.strip(),
            "intermediate_conclusion": final_conclusion,
            "is_final": True
        }


def display_streaming_reasoning(question):
    """
    显示流式推理过程
    
    Args:
        question: 问题
    """
    import time
    
    print(f"问题：{question}\n")
    print("="*70 + "\n")
    print("推理过程:\n")
    
    start_time = time.time()
    
    for step in streaming_chain_of_thought(question):
        # 显示步骤
        print(f"步骤 {step['step_number']}: {step['content']}")
        
        # 显示中间结论
        if not step['is_final']:
            print(f"  → 中间结论：{step['intermediate_conclusion']}\n")
        else:
            print(f"  → 最终结论：{step['intermediate_conclusion']}\n")
    
    total_time = (time.time() - start_time) * 1000
    print(f"\n[推理完成：总耗时 {total_time:.0f}ms]")


# 使用示例
if __name__ == "__main__":
    # 示例 1：数学推理
    print("="*70)
    print("示例 1：数学推理")
    print("="*70 + "\n")
    
    question1 = "小明有 5 个苹果，他给了小红 2 个，又买了 3 个，现在有几个苹果？"
    display_streaming_reasoning(question1)
    
    print("\n" + "="*70 + "\n")
    
    # 示例 2：逻辑推理
    print("示例 2：逻辑推理")
    print("="*70 + "\n")
    
    question2 = """
如果所有猫都怕水，而咪咪是一只猫，那么咪咪怕水吗？
如果咪咪实际上是一只机器猫，结论会变吗？
"""
    display_streaming_reasoning(question2)
    
    print("\n" + "="*70)
    print("\n关键观察:")
    print("1. 推理步骤逐条显示，用户可实时查看推理过程")
    print("2. 每个步骤都有中间结论，支持提前理解")
    print("3. 最终结论在所有步骤完成后得出")
    print("4. 用户可在推理过程中干预或修正前提")