### 实现 FastAPI 与 LangChain 的流式输出集成 在 Python 的 FastAPI 框架中，若需实现与 LangChain 的流式接口集成并返回流式响应，需要结合 `StreamingResponse` 类以及异步生成器函数。FastAPI 提供了对异步流式响应的支持，可以通过定义一个异步迭代器来逐步推送数据给客户端。 以下是一个完整的实现示例： ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_core.language_models import BaseChatModel from langchain_core.messages import HumanMessage, AIMessageChunk from langchain_openai import ChatOpenAI # 假设使用 OpenAI 作为后端模型 import asyncio app = FastAPI() # 假设已初始化一个支持流式的聊天模型，例如 OpenAI 模型 chat_model: BaseChatModel = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) @app.post("/stream") async def stream(): async def generate_stream(): # 使用 LangChain 的 astream 方法进行异步流式输出 async for chunk in chat_model.astream([HumanMessage(content="请介绍你自己")]): if isinstance(chunk, AIMessageChunk): yield f"data: {chunk.content}\n\n" # 按照 SSE 协议格式发送数据 return StreamingResponse(generate_stream(), media_type="text/event-stream") ``` 上述代码展示了如何通过 LangChain 的 `astream()` 方法获取异步流式输出，并将其封装为 FastAPI 的 `StreamingResponse` 对象返回给客户端[^2]。这种方式确保了服务器端能够逐步将 AI 输出内容推送给前端，从而实现类似“打字机”的实时渲染效果。 --- ### 异步流式处理的关键点 1. **使用 `astream()` 方法**：LangChain 的聊天模型提供了 `astream()` 接口用于异步流式处理，该方法返回一个异步迭代器，每次迭代返回一个包含部分输出的 `AIMessageChunk` 对象[^3]。 2. **封装为 `StreamingResponse`**：FastAPI 的 `StreamingResponse` 支持从异步生成器读取数据，并通过 HTTP 流式传输返回给客户端。 3. **遵循 Server-Sent Events (SSE) 协议**：在流式响应中，每个数据块应以 `data: ...` 格式返回，并以 `\n\n` 分隔，以便客户端正确解析。 --- ### 示例：自定义流式模型（可选） 若需进一步控制流式输出逻辑，可继承 `BaseChatModel` 并重写 `_stream()` 方法，逐字符或逐句生成输出。此方式适用于测试或特定业务场景： ```python from langchain_core.language_models import BaseChatModel from langchain_core.outputs import ChatGenerationChunk from langchain_core.messages import AIMessageChunk from typing import List, Optional, Any, Iterator class CustomStreamChatModel(BaseChatModel): model_name: str def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, **kwargs: Any, ) -> Iterator[ChatGenerationChunk]: last_message = messages[-1] for char in last_message.content: chunk = ChatGenerationChunk(message=AIMessageChunk(content=char)) yield chunk # 使用自定义模型 model = CustomStreamChatModel(model_name="custom_stream_model") for chunk in model.stream([HumanMessage(content="Hello, World!")]): print(chunk.message.content, end="") ``` 该示例展示了一个自定义的流式聊天模型，其 `_stream()` 方法按字符级别输出内容，可用于模拟流式行为[^3]。 --- ### 注意事项 - 确保使用的 LangChain 版本支持 `astream()` 方法。 - 若模型本身不支持流式输出，则无法实现逐字显示效果。 - 客户端需使用支持 `text/event-stream` 的方式接收响应，如 `EventSource` 或 Axios 流式请求。 ---