LangChain 教學：Models – 語言模型整合 → Streaming 模式

---

簡介

在使用大型語言模型（LLM）進行對話、文件摘要或程式碼產生時，即時回傳（Streaming） 已成為提升使用者體驗的關鍵技術。傳統的呼叫方式會等到模型全部產生完畢才返回結果，這在長篇回應或需要即時互動的場景下會造成明顯的延遲。LangChain 內建對多數主流 LLM（如 OpenAI、Anthropic、Cohere 等）的 Streaming 支援，讓開發者能夠在 收到第一個 token 時就開始顯示，大幅縮短感知等待時間。

本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，以及實務應用場景，完整介紹 LangChain 中的 Streaming 模式，幫助你快速在專案中導入即時回傳功能。

---

核心概念

1. 什麼是 Streaming？

Streaming 是指 LLM 在生成文字的過程中，逐步把 token（最小語意單位） 推送給呼叫端，而不是一次性等全部完成後才回傳。這類似於影片的串流播放，使用者可以「邊看邊下載」。

在 LangChain 中，Streaming 透過 streaming 參數或 AsyncIterator 物件實現。開發者只需要訂閱這個 iterator，即可在每個 token 產生時取得並即時呈現。

2. 為什麼要使用 Streaming？

優點	說明
降低感知延遲	使用者在等待過程中已看到部分回應，感覺更快。
提升互動性	可在聊天機器人、程式碼補全等需要即時回饋的情境中提供流暢體驗。
資源可控	允許在取得足夠資訊後即提前停止（如「只要前 3 行」），減少不必要的計算成本。

3. LangChain 中的 Streaming 實作方式

LangChain 主要提供兩種方式：

1. 同步呼叫 + 回傳 Generator
   使用 LLMChain 或 ChatOpenAI 時，將 streaming=True，回傳一個 generator，可用 for...of 迭代。

2. 非同步呼叫 + AsyncIterator
   若你的應用是基於 async/await（如 Node.js 的 Express、Fastify），可使用 await llm.stream(...) 取得 AsyncIterator。

下面分別示範這兩種用法。

---

程式碼範例

範例 1：基本的同步 Streaming（OpenAI GPT‑3.5）

import { ChatOpenAI } from "langchain/chat_models/openai";

// 建立模型，開啟 streaming
const chat = new ChatOpenAI({
  temperature: 0.7,
  modelName: "gpt-3.5-turbo",
  streaming: true,          // <─ 重要：啟用 streaming
});

async function run() {
  const response = await chat.call([
    { role: "system", content: "你是一位友善的助理。" },
    { role: "user", content: "請用繁體中文說明什麼是 Streaming。" },
  ]);

  // `response` 本身是個 generator，逐 token 取出
  for await (const chunk of response) {
    process.stdout.write(chunk.content); // 直接寫到終端機
  }
}

run();

重點：for await...of 能同時處理同步與非同步 generator，讓程式碼更簡潔。

---

範例 2：使用 LLMChain 搭配 PromptTemplate

import { OpenAI } from "langchain/llms/openai";
import { PromptTemplate } from "langchain/prompts";
import { LLMChain } from "langchain/chains";

const llm = new OpenAI({
  temperature: 0,
  streaming: true, // 啟用 streaming
});

const prompt = PromptTemplate.fromTemplate(
  "請將以下文字以條列式重點整理，並以繁體中文呈現：\n\n{input}"
);

const chain = new LLMChain({ llm, prompt });

async function summarize(text) {
  const iterator = await chain.run({ input: text });
  // `iterator` 為 generator，逐 token 輸出
  for await (const chunk of iterator) {
    process.stdout.write(chunk);
  }
}

// 測試文字
const article = "LangChain 是一個用於建構 LLM 應用的框架，提供了模型、記憶、工具等模組化組件。...";
summarize(article);

此範例展示 PromptTemplate 與 LLMChain 的結合，讓開發者只需關注 Prompt 與輸入資料，Streaming 的細節交給 LangChain 處理。

---

範例 3：非同步 Streaming（Node.js + Express）

import express from "express";
import { ChatOpenAI } from "langchain/chat_models/openai";

const app = express();
app.use(express.json());

const chat = new ChatOpenAI({
  modelName: "gpt-4",
  temperature: 0.6,
  streaming: true,
});

app.post("/chat", async (req, res) => {
  const { messages } = req.body; // [{role: "user", content: "..."}]

  // 設定 response 為 SSE (Server‑Sent Events)
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.flushHeaders();

  const iterator = await chat.call(messages);

  for await (const chunk of iterator) {
    // 每個 token 包成 SSE 格式傳送
    res.write(`data: ${JSON.stringify(chunk)}\n\n`);
  }

  // 完成後結束 SSE 連線
  res.write("event: done\ndata: {}\n\n");
  res.end();
});

app.listen(3000, () => console.log("Server running on http://localhost:3000"));

這裡使用 Server‑Sent Events 把每個 token 即時推送到前端，前端只要監聽 message 事件即可逐字顯示回應，實現流暢的聊天體驗。

---

範例 4：自訂停止條件（只要前 5 行就結束）

import { ChatOpenAI } from "langchain/chat_models/openai";

const chat = new ChatOpenAI({
  modelName: "gpt-3.5-turbo",
  streaming: true,
});

async function generateLimitedResponse() {
  const iterator = await chat.call([
    { role: "system", content: "你是一位技術部落客。" },
    { role: "user", content: "寫一段 1000 字的文章。" },
  ]);

  let lineCount = 0;
  for await (const chunk of iterator) {
    process.stdout.write(chunk.content);
    if (chunk.content.includes("\n")) lineCount++;
    if (lineCount >= 5) break; // 取得 5 行後提前結束
  }
}

generateLimitedResponse();

透過 手動 break，可以在取得足夠資訊後即時停止迭代，降低 API 使用成本。

---

範例 5：結合 VectorStore 做「即時檢索 + 生成」的混合模式

import { ChatOpenAI } from "langchain/chat_models/openai";
import { OpenAIEmbeddings } from "langchain/embeddings/openai";
import { MemoryVectorStore } from "langchain/vectorstores/memory";
import { ConversationalRetrievalChain } from "langchain/chains";

const llm = new ChatOpenAI({ streaming: true, temperature: 0 });
const embeddings = new OpenAIEmbeddings();
const vectorStore = await MemoryVectorStore.fromTexts(
  ["LangChain 是什麼？", "Streaming 可以降低延遲。", "LangChain 支援多種模型。"],
  [],
  embeddings
);

const chain = ConversationalRetrievalChain.fromLLM({
  llm,
  retriever: vectorStore.asRetriever(),
});

async function ask(question) {
  const iterator = await chain.call({ question, chat_history: [] });
  for await (const chunk of iterator) {
    process.stdout.write(chunk);
  }
}

ask("什麼是 LangChain 的 Streaming 模式？");

此範例示範 檢索增強生成（RAG） 與 Streaming 的結合，使用者在等待生成答案的同時，也能即時看到來自向量資料庫的相關資訊。

---

常見陷阱與最佳實踐

陷阱	說明	解決方案 / 最佳實踐
忘記設定 streaming: true	預設是同步呼叫，會一次回傳完整文字，失去即時效果。	建立模型時務必加入 streaming: true，或在 call() 時傳入 { streaming: true }。
在同步環境使用 for await...of	若執行環境不支援 async iterator，會拋出錯誤。	確認 Node 版本 ≥ 14，或改用回呼（callback）方式處理。
過度頻繁的 break	手動提前結束迭代時，若在 token 邊界斷開，可能產生不完整的句子。	建議使用 token 數量 或 句點 判斷停止點，或利用模型的 stop 參數。
忘記釋放資源	長時間開啟 SSE 但未正確關閉，會佔用伺服器連線資源。	在迭代結束或發生例外時，務必 res.end()，或使用 try…finally 包裝迭代。
前端未正確處理 SSE	前端若只接收一次 message，會看不到持續更新。	前端使用 EventSource 並在 onmessage 中累加顯示內容。

最佳實踐總結：

1. 統一使用 async iterator：無論同步或非同步，都以 for await...of 為主，保持程式碼一致性。
2. 設定合理的 maxTokens：避免模型產生過長輸出，減少不必要的 streaming 時間。
3. 使用 stop 參數：在 Prompt 中加入明確的停止字元，讓模型自行結束，降低手動 break 的風險。
4. 前端顯示優化：使用「打字機」效果或「逐字顯示」動畫，提升使用者感受。
5. 錯誤處理：將迭代包在 try…catch，確保任何網路或模型錯誤都能優雅回報。

---

實際應用場景

場景	為何需要 Streaming	示範實作概念
聊天機器人	使用者期待即時回應，尤其在手機或網頁上，等待整段文字會感到卡頓。	使用 SSE 或 WebSocket 把 token 逐一推送前端。
程式碼補全	IDE 需要在開發者鍵入時即時顯示建議，延遲會干擾開發流程。	在 VS Code Extension 中，利用 for await...of 把每個 token 作為補全項目送出。
長文件摘要	摘要可能需要上百個 token，使用者可先看到前半段摘要，決定是否繼續閱讀。	結合 VectorStore 做 RAG，先返回最相關的段落，再持續生成完整摘要。
即時翻譯	翻譯服務在會議或直播中需要毫秒級回應。	把原文分段送入 LLM，使用 Streaming 回傳翻譯結果。
教育平台互動題目	學生提問後希望立即看到解說，避免等待過長造成注意力流失。	透過 Streaming 把解說一步步呈現，並在每一步加入提示（如「下一步」按鈕）。

---

總結

Streaming 是 提升大型語言模型交互體驗的關鍵技術，在 LangChain 中只需要簡單的 streaming: true 配置，即可將 token 逐步推送給前端或其他消費者。本文從概念、程式碼範例、常見陷阱與最佳實踐，以及多種實務應用場景，完整說明了如何在 Node.js / JavaScript 生態中使用 LangChain 的 Streaming 功能。

重點回顧

• 設定 streaming: true 才會得到 async iterator。
• 使用 for await...of 迭代 token，搭配 SSE / WebSocket 即可在前端即時顯示。
• 透過 stop、maxTokens 與自訂的 提前停止條件，可以控制成本與回應長度。
• 結合向量檢索（RAG）時，同樣支援 Streaming，讓檢索與生成同步進行。

掌握了這些要點，你就能在聊天機器人、程式碼補全、即時翻譯等各種情境下，打造出 流暢、即時且具成本效益 的 LLM 應用。祝開發順利，期待看到你用 LangChain 搭建的精彩作品！ 🚀