企业官网建设流程全解析

1. 为什么流式输出和结构化输出总是打架

做过大模型应用的人多半踩过这个坑：前端要打字机效果，后端要拿到能入库、能调函数、能渲染成卡片的结构化数据，这两件事天然是矛盾的。SSE 流式输出的本质是"边生成边推送"，模型吐一个字你推一个字，用户看着爽；而结构化输出要求的是"完整、合法、可解析"，你得等整个 JSON 闭合之后才能json.loads。一个要快，一个要全，硬凑在一起就会出现那种经典现象——流到一半断了，前端收到半截 JSON，解析直接炸。

我最早做对话产品的时候，图省事，直接让模型输出 JSON，然后前端拿TextDecoder一点点拼，拼完再JSON.parse。小 demo 跑得挺欢，一上量就原形毕露：模型偶尔在 JSON 外面加一句"好的，以下是结果："，偶尔少个右括号，偶尔把中文引号当英文引号用。后来换成 LangChain 的 OutputParser 体系，才算把这件事理顺。再往后做 Agent，需要模型在对话中途调用工具，又引入了 ToolCall，这时候流式和结构化的矛盾更尖锐了——工具调用的参数也是结构化的，而且往往在流式过程中就要决定调不调、调哪个。

这篇就把我这几年的实战经验摊开讲。核心围绕三块：LangChain 的三大 OutputParser（PydanticOutputParser、JsonOutputParser、StructuredOutputParser）到底怎么选、怎么用；ToolCall 在流式场景下怎么和 Parser 配合；SSE 这条链路从后端 FastAPI 到前端解析，中间有哪些必须处理的边界情况。适合已经能跑通 LangChain 基础链、但一碰到"流式 + 结构化"就卡壳的同学，也适合正在做 Agent 二次开发、需要把工具调用和流式输出缝合起来的人。

先说结论，省得你看到后面才反应过来：流式和结构化不是二选一，而是分层处理。流式负责传输层和用户体验，结构化负责业务层和数据契约，中间靠 Parser 做转换，靠 ToolCall 做决策。想清楚这个分层，后面所有问题都有解。

2. LangChain 三大 OutputParser 到底怎么选

2.1 PydanticOutputParser：契约最严，但最不抗造

PydanticOutputParser 是我用得最多、也最推荐新手先上手的一个。它的逻辑很直白：你定义一个 Pydantic 模型，它自动生成一段格式说明塞进 prompt，模型按说明输出 JSON，Parser 再把它反序列化成模型实例。好处是类型安全，字段缺了、类型错了，Pydantic 直接给你报错，不会让脏数据流到下游。

from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class ProductInfo(BaseModel): name: str = Field(description="产品名称") price: float = Field(description="价格，单位元") tags: list[str] = Field(description="标签列表") parser = PydanticOutputParser(pydantic_object=ProductInfo) format_instructions = parser.get_format_instructions()

get_format_instructions()返回的那段文字很关键，它会告诉模型"输出必须是合法 JSON，字段有哪些，类型是什么"。我一般会把它拼进 system prompt 的末尾，而不是 user message 里，因为 system 的约束力更强，模型更不容易忽略。

但它的短板也很明显：对模型输出的容错几乎为零。模型多写一个逗号、少一个引号，Parser 就抛OutputParserException。我实测下来，GPT-4 级别的模型按格式输出成功率能到 95% 以上，但小模型或者中文场景下，失败率会明显上升。所以用它的时候，必须配重试机制，这个后面第 4 节会详细讲。

2.2 JsonOutputParser：灵活，适合流式场景

JsonOutputParser 是 PydanticOutputParser 的"宽松版"。它不强制你定义模型，输出就是个 dict，而且它有个杀手锏——支持流式增量解析。也就是说，模型还在吐 JSON 的时候，你就能拿到已经解析出来的部分字段。

from langchain_core.output_parsers import JsonOutputParser parser = JsonOutputParser() chain = prompt | llm | parser for chunk in chain.stream({"query": "..."}): print(chunk) # 逐步拿到部分解析的 dict

这个特性在流式场景下太重要了。比如你要做一个"边生成边渲染卡片"的功能，用户不用等整个 JSON 出来，字段一到位就能先渲染。我做过一个商品推荐场景，模型先吐name，前端立刻显示标题，再吐price，价格区域补上，体验比等全量 JSON 好太多。

代价是它不做类型校验。price字段模型给你返回个字符串"99.9"，它也照收，你得自己在业务层做转换。所以我的习惯是：流式阶段用 JsonOutputParser 拿增量，流结束后再用 Pydantic 模型做一次严格校验，两层保险。

2.3 StructuredOutputParser：多字段场景的折中方案

StructuredOutputParser 介于两者之间。它不依赖 Pydantic 模型，而是用ResponseSchema手动定义字段，输出也是 dict。适合那种字段不多、又不想引入 Pydantic 依赖的场景。

from langchain.output_parsers import StructuredOutputParser, ResponseSchema schemas = [ ResponseSchema(name="summary", description="摘要"), ResponseSchema(name="sentiment", description="情感倾向，positive/negative/neutral"), ] parser = StructuredOutputParser.from_response_schemas(schemas)

说实话，现在新项目我基本不用它了，因为 Pydantic 的表达能力更强，JsonOutputParser 的流式支持更好，它夹在中间有点尴尬。但如果你维护的是老代码，或者团队不想引入 Pydantic，它仍然是个稳妥选择。

2.4 三者对比与选型建议

选型逻辑我总结成一句话：要类型安全选 Pydantic，要流式选 Json，要轻量选 Structured。实际项目里经常是组合使用，比如 Agent 的工具参数用 Pydantic 校验，最终回复用 Json 流式推给前端。

3. ToolCall 与流式输出的缝合实战

3.1 ToolCall 的本质是"结构化决策"

很多人把 ToolCall 想得很玄，其实它的本质就是：模型输出一段结构化的 JSON，描述"我要调用哪个函数、参数是什么"，框架解析这段 JSON 后去执行真正的函数。所以 ToolCall 天然就是结构化输出的一种，只不过它的 schema 是工具的签名。

from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" return f"{city}今天晴，25度" llm_with_tools = llm.bind_tools([get_weather]) response = llm_with_tools.invoke("北京天气怎么样") # response.tool_calls 里就是结构化的调用意图

bind_tools之后，模型返回的AIMessage里会带tool_calls字段，这就是结构化的调用意图。注意，这时候模型并没有真的调用函数，它只是"说"要调用，真正的执行要你自己写循环去处理。

3.2 流式场景下 ToolCall 的特殊处理

流式 + ToolCall 的坑在于：工具调用的参数是分片到达的。模型可能先吐{"city": "北，再吐京"}，你不能拿到第一片就去执行。LangChain 的AIMessageChunk会把tool_call_chunks累积起来，你需要等流结束或者检测到完整的调用意图再执行。

full_message = None for chunk in llm_with_tools.stream("北京天气怎么样"): full_message = chunk if full_message is None else full_message + chunk if full_message.tool_calls: for call in full_message.tool_calls: result = get_weather.invoke(call["args"])

这里有个细节：AIMessageChunk支持+运算符合并，这是 LangChain 设计得很巧妙的地方。合并之后tool_calls才是完整的。我见过有人直接在循环里判断chunk.tool_calls，结果拿到的是残缺参数，调函数直接报错。

3.3 流式输出与工具调用的时序问题

一个完整的 Agent 回合通常是：模型思考 → 决定调工具 → 工具执行 → 模型基于结果继续生成。如果全程流式，用户会看到"思考中"→"调用工具"→"工具结果"→"最终回答"这几个阶段。我的做法是分段流式：工具调用阶段不推给用户，只推一个"正在查询"的状态；最终回答阶段才真正流式推文本。

这样做的理由是，工具调用的 JSON 对用户没有意义，推过去只会让界面乱。而最终回答是用户真正关心的，流式体验最好。这个取舍在 Agent 产品里几乎是标配。

4. SSE 链路从后端到前端的完整实现

4.1 FastAPI 侧：封装 SSE 流式接口

后端用 FastAPI 做 SSE 是最顺手的，StreamingResponse配合生成器就能搞定。但要注意几个细节：响应头必须设text/event-stream，要禁用缓冲，还要处理客户端断开。

from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() async def event_generator(query: str): async for chunk in chain.astream({"query": query}): data = json.dumps({"content": chunk}, ensure_ascii=False) yield f"data: {data}\n\n" yield "data: [DONE]\n\n" @app.get("/chat") async def chat(query: str): return StreamingResponse( event_generator(query), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, )

X-Accel-Buffering: no这个头很关键，如果你前面挂了 Nginx，不加这个它会把流缓冲起来，用户看到的就不是打字机效果，而是等半天一次性全出来。我第一次部署就栽在这，本地好好的，一上服务器就"卡住"，排查了半天才发现是 Nginx 的锅。

4.2 前端侧：封装 SSE 解析逻辑

前端解析 SSE 有个经典坑：一个 chunk 里可能包含多条消息，一条消息也可能跨多个 chunk。所以不能简单地按 chunk 切分，必须维护一个缓冲区，按\n\n分隔符切。

async function consumeSSE(url, onMessage) { const response = await fetch(url); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const parts = buffer.split("\n\n"); buffer = parts.pop(); // 最后一段可能不完整，留到下次 for (const part of parts) { if (part.startsWith("data: ")) { const data = part.slice(6); if (data === "[DONE]") return; onMessage(JSON.parse(data)); } } } }

buffer = parts.pop()这行是整个解析的核心。它把最后一段不完整的消息留在缓冲区，等下一个 chunk 来了再拼。我见过太多人直接split完就遍历，结果遇到跨 chunk 的消息就解析失败，报那种Unexpected end of JSON input的错。

4.3 处理 idle timeout 与断流重连

热词里那个stream disconnected before completion: idle timeout waiting for sse是 SSE 最常见的故障。原因通常是：模型思考时间太长，中间没有数据推送，网关或浏览器判定连接空闲，主动断开。

解决办法有两个方向。一是心跳保活，在生成器里定期推一个空注释行：

async def event_generator(query: str): # 启动一个心跳任务，每 15 秒推一次 yield ": keep-alive\n\n" async for chunk in chain.astream({"query": query}): ...

以:开头的行是 SSE 的注释，前端会忽略，但能保持连接活跃。二是前端重连，检测到断流后带上已接收的内容重新请求，让模型从断点续写。这个实现复杂一些，但体验更好。我的经验是，心跳保活能解决 90% 的 idle timeout，剩下的靠重连兜底。

5. 常见问题与排查技巧实录

5.1 Parser 报错速查表

5.2 我的避坑心得

第一个心得：永远不要相信模型会严格按格式输出。哪怕你 prompt 写得再清楚，也要在 Parser 外面包一层 try-except，失败就重试或者降级。我现在的标准做法是parser.with_retry(stop_after_attempt=3)，三次还失败就返回一个兜底结构，绝不让异常穿透到用户。

第二个心得：流式场景下，结构化字段要分优先级。不是所有字段都需要实时渲染，把用户最关心的字段放在 JSON 前面，模型先吐出来，前端先渲染。比如商品卡片，name和price放前面，description放后面，体验会好很多。

第三个心得：调试 SSE 一定要用 curl。浏览器和前端框架会帮你做很多隐式处理，出问题时你根本不知道原始流长什么样。curl -N http://localhost:8000/chat?query=test加上-N禁用缓冲，能看到最原始的字节流，排查问题快得多。

5.3 性能与稳定性建议

流式接口的稳定性，很大程度取决于超时设置。我的经验值是：单次生成超过 60 秒就该考虑拆分任务，SSE 连接超过 5 分钟就该主动断开让前端重连。另外，astream比stream更适合 FastAPI 这种异步框架，别用错。

还有一个容易被忽略的点：并发流式请求的资源占用。每个 SSE 连接都会占一个协程，如果同时有几百个连接，内存和文件描述符都会吃紧。生产环境建议加连接数限制，或者用队列把请求排队处理。

6. 一个完整的可复现示例

把前面的东西串起来，给一个能直接跑的完整例子。后端 FastAPI + LangChain，前端原生 JS，实现"流式输出 + 结构化字段增量渲染"。

后端核心逻辑：

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser parser = JsonOutputParser() prompt = ChatPromptTemplate.from_template( "根据用户需求生成商品信息，输出 JSON。\n{format_instructions}\n需求：{query}" ).partial(format_instructions=parser.get_format_instructions()) chain = prompt | llm | parser async def event_generator(query: str): yield ": keep-alive\n\n" async for partial in chain.astream({"query": query}): yield f"data: {json.dumps(partial, ensure_ascii=False)}\n\n" yield "data: [DONE]\n\n"

前端拿到增量 dict 后，按字段是否存在决定渲染哪块 UI。name到了就显示标题，price到了就显示价格，不用等全量。

这套方案我在两个项目里用过，一个商品推荐，一个工单分类，稳定性都不错。关键就是把流式、结构化、工具调用这三层分清楚，各司其职，别让它们互相干扰。踩过的坑基本都在上面了，剩下的就是根据你的业务场景微调字段优先级和超时参数。