LangChain 流式输出实现与前端集成完整教程

我第一次在生产环境对接流式输出时，踩了整整三天的坑。那时候团队要做一个 AI 写作助手，用户抱怨打字效果太僵硬，不像真正的对话体验。后来我仔细算了算成本账，才发现问题的根源不在前端，而在 API 调用方式和计费策略上。

先算一笔真实费用：为什么流式输出值得优化

我们拿 2026 年主流模型的 output 价格做个对比：GPT-4.1 是 $8/MTok、Claude Sonnet 4.5 是 $15/MTok、Gemini 2.5 Flash 是 $2.50/MTok、DeepSeek V3.2 是 $0.42/MTok。假设你每月处理 100 万 output token，用 立即注册 的 HolySheep API（汇率 ¥1=$1 无损，官方汇率是 ¥7.3=$1），对比一下差距：

• GPT-4.1：使用 HolySheep 只需 ¥8，官方渠道要 ¥58.4，节省 86%
• Claude Sonnet 4.5：HolySheep ¥15，官方渠道 ¥109.5，节省 86%
• Gemini 2.5 Flash：HolySheep ¥2.5，官方渠道 ¥18.25，节省 86%
• DeepSeek V3.2：HolySheep ¥0.42，官方渠道 ¥3.07，节省 86%

如果你的产品月调用量达到 1000 万 token，光是 GPT-4.1 这一款模型就能省下 ¥58,400。这个数字让我意识到，中转 API 的价值不只是「绕过限制」，更重要的是让国内开发者用人民币直接结算，省去换汇麻烦和汇损。

LangChain 流式输出核心原理

LangChain 的流式输出本质上依赖 Python 的生成器（Generator）机制。当你调用 stream() 方法时，模型返回的不是完整响应，而是一个迭代器，每个元素是一个 token 片段。这样前端可以实时渲染，用户看到的是逐字出现的效果，而不是等待 5 秒后一次性显示。

# 标准 LangChain 流式调用示例（使用 HolySheep API）
from langchain_openai import ChatOpenAI

配置 HolySheep API 端点，国内延迟 <50ms
llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为你的 HolySheep Key
    streaming=True  # 关键参数：开启流式输出
)

流式调用示例
for chunk in llm.stream("请用三句话解释什么是量子计算"):
    print(chunk.content, end="", flush=True)

我自己项目里用的方案是把流式输出和 SSE（Server-Sent Events）结合。FastAPI 后端接收请求后，调用 LangChain stream 方法，然后把每个 chunk 通过 ASGI 的 WebSocket 或者 SSE 推送给前端。

# FastAPI + LangChain 流式输出后端完整实现
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import asyncio

app = FastAPI()

初始化 HolySheep API 客户端
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    streaming=True
)

@app.post("/chat/stream")
async def chat_stream(request: Request):
    """
    流式聊天接口，返回 SSE 格式数据
    前端通过 EventSource 或 fetch 接收
    """
    body = await request.json()
    user_message = body.get("message", "")
    
    async def event_generator():
        # 调用 LangChain 流式方法
        async for chunk in llm.astream([HumanMessage(content=user_message)]):
            # 转换为 SSE 格式
            yield f"data: {chunk.content}\n\n"
            await asyncio.sleep(0.01)  # 防止过快推送
        
        # 发送结束标记
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

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

前端集成：三种主流实现方案

方案一：原生 Fetch + ReadableStream（推荐）

这是最轻量的方案，兼容所有现代浏览器，不需要任何第三方依赖。我在公司的知识库问答产品里用的就是这个，延迟体感非常流畅。

<!-- HTML 结构 -->
<div id="chat-container">
    <div id="messages"></div>
    <textarea id="user-input" placeholder="输入你的问题..."></textarea>
    <button onclick="sendMessage()">发送</button>
</div>

<script>
async function sendMessage() {
    const input = document.getElementById('user-input');
    const messages = document.getElementById('messages');
    const userMessage = input.value;
    
    // 显示用户消息
    messages.innerHTML += <div class="user">${userMessage}</div>;
    
    // 创建 AI 响应容器
    const aiDiv = document.createElement('div');
    aiDiv.className = 'ai';
    aiDiv.textContent = '思考中...';
    messages.appendChild(aiDiv);
    
    input.value = '';
    
    // 调用流式接口
    const response = await fetch('https://your-api-domain.com/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ message: userMessage })
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    // 逐块读取并渲染
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        
        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');
        
        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') continue;
                
                // 追加内容，模拟打字机效果
                aiDiv.textContent += data;
            }
        }
    }
}
</script>

方案二：Vue 3 组件封装（适合 SPA 项目）

<!-- Vue 3 流式聊天组件 -->
<template>
  <div class="chat-window">
    <div v-for="(msg, idx) in messages" :key="idx" :class="msg.role">
      {{ msg.content }}
    </div>
    <div v-if="streaming" class="ai typing">{{ streamingContent }}<span class="cursor">█</span></div>
  </div>
  <div class="input-area">
    <textarea v-model="inputText" @keydown.enter.exact.prevent="sendMessage"></textarea>
    <button @click="sendMessage" :disabled="loading">
      {{ loading ? '发送中' : '发送' }}
    </button>
  </div>
</template>

<script setup>
import { ref } from 'vue';

const inputText = ref('');
const messages = ref([]);
const loading = ref(false);
const streaming = ref(false);
const streamingContent = ref('');

async function sendMessage() {
    if (!inputText.value.trim() || loading.value) return;
    
    const userMsg = inputText.value;
    messages.value.push({ role: 'user', content: userMsg });
    inputText.value = '';
    
    loading.value = true;
    streaming.value = true;
    streamingContent.value = '';
    
    try {
        const response = await fetch('https://your-api-domain.com/chat/stream', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ message: userMsg })
        });
        
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        
        while (true) {
            const { done, value } = await reader.read();
            if (done) break;
            
            const chunk = decoder.decode(value);
            chunk.split('\n').forEach(line => {
                if (line.startsWith('data: ') && line.slice(6) !== '[DONE]') {
                    streamingContent.value += line.slice(6);
                }
            });
        }
        
        // 流式结束，保存完整消息
        messages.value.push({ role: 'ai', content: streamingContent.value });
    } catch (error) {
        console.error('流式调用失败:', error);
        messages.value.push({ role: 'ai', content: '抱歉，服务暂时不可用。' });
    } finally {
        loading.value = false;
        streaming.value = false;
    }
}
</script>

方案三：React + useSSE Hook（适合 Next.js 项目）

import { useState, useCallback, useRef } from 'react';

// 自定义 Hook：管理 SSE 流式请求
function useStreamChat() {
    const [messages, setMessages] = useState([]);
    const [streamingText, setStreamingText] = useState('');
    const [isStreaming, setIsStreaming] = useState(false);
    const abortControllerRef = useRef(null);

    const sendMessage = useCallback(async (content) => {
        // 取消之前的请求
        if (abortControllerRef.current) {
            abortControllerRef.current.abort();
        }
        
        abortControllerRef.current = new AbortController();
        
        setMessages(prev => [...prev, { role: 'user', content }]);
        setIsStreaming(true);
        setStreamingText('');

        try {
            const response = await fetch('https://your-api-domain.com/chat/stream', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({ message: content }),
                signal: abortControllerRef.current.signal
            });

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

            while (true) {
                const { done, value } = await reader.read();
                if (done) break;

                const chunk = decoder.decode(value);
                chunk.split('\n').forEach(line => {
                    if (line.startsWith('data: ') && line.slice(6) !== '[DONE]') {
                        setStreamingText(prev => prev + line.slice(6));
                    }
                });
            }

            // 保存到消息历史
            setMessages(prev => [...prev, { role: 'assistant', content: streamingText }]);
            setStreamingText('');
        } catch (error) {
            if (error.name !== 'AbortError') {
                console.error('请求失败:', error);
            }
        } finally {
            setIsStreaming(false);
        }
    }, [streamingText]);

    return { messages, streamingText, isStreaming, sendMessage };
}

export default useStreamChat;

HolySheep API 在 LangChain 中的高级配置

我在多个项目里总结出一套高效的配置模板，特别适合需要同时调用多个模型做对比测试的场景。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
from typing import Dict, Optional
import os

class MultiModelClient:
    """HolySheep 多模型统一客户端"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        
    def get_client(self, provider: str, model: str) -> ChatOpenAI:
        """根据提供商和模型返回对应客户端"""
        config = {
            "openai": {
                "class": ChatOpenAI,
                "default_model": "gpt-4.1"
            },
            "anthropic": {
                "class": ChatOpenAI,  # LangChain 统一接口
                "default_model": "claude-sonnet-4.5"
            },
            "google": {
                "class": ChatOpenAI,  # 使用 OpenAI 兼容接口
                "default_model": "gemini-2.0-flash"
            },
            "deepseek": {
                "class": ChatOpenAI,
                "default_model": "deepseek-v3.2"
            }
        }
        
        client_config = config.get(provider, config["openai"])
        client_class = client_config["class"]
        
        return client_class(
            model=model or client_config["default_model"],
            base_url=self.base_url,
            api_key=self.api_key,
            streaming=True
        )
    
    async def stream_chat(self, provider: str, model: str, prompt: str) -> str:
        """统一的流式聊天接口"""
        client = self.get_client(provider, model)
        
        full_response = ""
        async for chunk in client.astream(prompt):
            full_response += chunk.content
            # 这里可以添加实时回调，比如推送到前端
            yield chunk.content

使用示例
if __name__ == "__main__":
    client = MultiModelClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # 测试 DeepSeek 模型（性价比最高）
    print("DeepSeek V3.2 输出：")
    for token in client.stream_chat("deepseek", "deepseek-v3.2", "解释什么是注意力机制"):
        print(token, end="", flush=True)

常见报错排查

我把过去一年踩过的坑整理成这份清单，覆盖了 90% 以上的流式输出异常场景。

报错 1：streaming=True 但返回完整响应

错误信息：RuntimeWarning: coroutine 'ChatOpenAI.stream' was never awaited 或者前端收到的是一次性完整数据而非分块。

根本原因：没有正确使用异步迭代器。LangChain 的 stream 方法返回的是生成器对象，必须用 for ... in 遍历，不能直接 .stream() 调用。

# ❌ 错误写法
response = llm.stream("hello")
print(response)  # 返回的是生成器对象，不是数据

✅ 正确写法
for chunk in llm.stream("hello"):
    print(chunk.content)

✅ 异步场景
async for chunk in llm.astream("hello"):
    print(chunk.content)

报错 2：CORS 跨域问题导致前端无法接收流

错误信息：Access to fetch at 'https://your-api.com' from origin 'http://localhost:3000' has been blocked by CORS policy

根本原因：FastAPI 默认没有配置 CORS 头，流式响应被浏览器拦截。

# ✅ 正确配置 FastAPI CORS
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://your-frontend.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

特别针对 SSE 流式响应，确保 headers 正确
@app.post("/chat/stream")
async def chat_stream(request: Request):
    async def generate():
        async for chunk in llm.astream([HumanMessage(content="test")]):
            yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type, Authorization",
            "X-Accel-Buffering": "no"  # 禁用 Nginx 缓冲
        }
    )

报错 3：Nginx 代理导致流式输出延迟堆积

错误信息：前端显示字符不是实时出现，而是等几秒后突然一次性显示所有内容。

根本原因：Nginx 默认会缓冲响应体以提高性能，这对于 SSE 流式输出是致命的。

# ✅ Nginx 配置修改
server {
    listen 80;
    server_name your-api-domain.com;
    
    location /chat/stream {
        # 禁用代理缓冲，确保实时推送
        proxy_buffering off;
        proxy_cache off;
        
        # 设置正确的超时时间
        proxy_read_timeout 300s;
        proxy_connect_timeout 75s;
        
        proxy_pass http://localhost:8000;
        
        # 关键配置：关闭缓冲
        proxy_request_buffering off;
        
        # 添加必要的响应头
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        
        # 确保 Transfer-Encoding 为 chunked
        chunked_transfer_encoding on;
    }
}

报错 4：API Key 无效或余额不足

错误信息：AuthenticationError: Incorrect API key provided 或 RateLimitError: You have exceeded your monthly rate limit

根本原因：HolySheep 使用的是独立 API Key，与官方 OpenAI 密钥不通用。如果你在代码里配置了错误的 base_url，也会导致认证失败。

# ✅ 完整的环境变量配置示例
import os
from langchain_openai import ChatOpenAI

从环境变量读取，永不在代码中硬编码
api_key = os.environ.get("HOLYSHEEP_API_KEY")

if not api_key:
    raise ValueError("请设置 HOLYSHEEP_API_KEY 环境变量")

llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",  # 确认使用 HolySheep 端点
    api_key=api_key,
    streaming=True,
    max_retries=3,
    timeout=60
)

测试连接
try:
    response = next(llm.stream("hello"))
    print(f"连接成功: {response.content}")
except Exception as e:
    print(f"连接失败: {e}")
    # 推荐登录 HolySheep 检查余额和 Key

报错 5：Vue/React 前端 SSE 事件解析不完整

错误信息：控制台显示收到数据，但页面上只显示部分内容，或者出现乱码。

根本原因：SSE 的 data 字段可能包含多行内容，或者 chunk 被 TCP 分片传输，需要完整的行解析逻辑。

// ✅ 健壮的 SSE 解析函数
function parseSSEStream(response) {
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    
    return new ReadableStream({
        async start(controller) {
            while (true) {
                const { done, value } = await reader.read();
                if (done) {
                    controller.close();
                    break;
                }
                
                buffer += decoder.decode(value, { stream: true });
                const lines = buffer.split('\n');
                buffer = lines.pop() || ''; // 保留不完整的行
                
                for (const line of lines) {
                    if (line.startsWith('data: ')) {
                        const data = line.slice(6).trim();
                        if (data === '[DONE]') {
                            controller.close();
                            return;
                        }
                        // 将解析后的内容传递给消费者
                        controller.enqueue(data);
                    }
                }
            }
        }
    });
}

// 使用示例
async function chat(message) {
    const response = await fetch('/chat/stream', {
        method: 'POST',
        body: JSON.stringify({ message }),
        headers: { 'Content-Type': 'application/json' }
    });
    
    const stream = parseSSEStream(response);
    const reader = stream.getReader();
    
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        // 追加到页面显示
        document.getElementById('output').textContent += value;
    }
}

性能优化实战经验

我负责的 AI 客服系统经历过一次重大重构，从同步调用改成流式输出后，用户满意度从 72% 提升到 89%。有几个关键优化点值得分享：

• 首字节时间（TTFB）优化：HolySheep 的国内节点延迟 <50ms，配合 FastAPI 的异步流式响应，平均首字节时间可以压到 200ms 以内
• 连接复用：不要每次请求都创建新连接，使用 httpx 或 aiohttp 的连接池
• 分块大小控制：如果模型输出太快（DeepSeek V3.2 这种），可以在后端加 asyncio.sleep(0.01) 防止前端渲染跟不上
• 错误重试：流式中断很常见，建议前端实现自动重连逻辑

总结

LangChain 的流式输出实现本身并不复杂，核心就三个点：后端生成器、前端 SSE 解析、以及正确的错误处理。选择 HolySheep API 作为中转，不只是省去换汇麻烦，更重要的是国内直连的低延迟和人民币结算的便利性。¥1=$1 的汇率相比官方 ¥7.3=$1，长期来看能节省一大笔成本。

建议新项目从 DeepSeek V3.2 开始接入，$0.42/MTok 的价格非常有竞争力，适合做内部工具和 POC。等产品验证后再切换到 GPT-4.1 或 Claude Sonnet 4.5 做生产级输出。

👉 免费注册 HolySheep AI，获取首月赠额度