在LLM应用开发里最磨人的一件事就是把模型吐出来的一堆文本变成后端能直接消费的数据结构。做过这类项目的同学应该都有印象你让模型返回JSON它偶尔给你讲一段话你让模型按照固定字段输出它给你来个null值更别提转义混乱、字段名变形这些日常问题。折腾一圈下来你可能会觉得与其和文本斗智斗勇不如直接给模型一张“表格”让它填。“表格”其实就是结构化输出的一种形象说法。LLM本身只会生成文本流但我们可以通过Tool Call工具调用这种机制让模型以“调用工具”的形式返回严格符合参数定义的JSON。而withStructuredOutput这个接口把请求参数注入、JSON校验、失败重试这些脏活全部封装成一两行代码直接拿到一个类型明确的模型对象。这篇文章我会从原理讲到落地重点拆解withStructuredOutput如何一键打通Tool Call再结合流式输出场景最后把结果安全地写进MySQL。适合正在做LLM数据抽取、Agent工具调用、或者被“模型输出JSON解析”折磨过的开发者参考。1. 从文本到Schema为什么要放弃“零散JSON”方案在进入withStructuredOutput之前先捋清楚一个底层问题LLM输出的结构化到底难在哪里为什么简单的“加个prompt让它返回JSON”不够用。1.1 原生JSON输出的三个不稳定性不少初学LLM开发的同学第一步都会这么写prompt 请从这段文本中提取信息以JSON格式返回{...} response llm.invoke(prompt) data json.loads(response.content)问题在于模型返回的内容本质是“一张字条”不是“一份合同”。你要求它返回JSON它可能乖乖给JSON也可能在JSON前面加一句“好的以下是提取结果”。更常见的情况是字段名飘忽不定userId可能变成user_iduser-id甚至用户ID。类型不严格数字字段可能返回字符串18也可能返回18。转义损坏内容里包含引号、换行符时json.loads直接抛异常。多余内容混入模型在JSON后面继续写了一段总结。这些问题不是靠“把prompt写得更清楚”就能彻底解决的。你写“只输出JSON”它偶尔会遵守你写“不要输出无关内容”它偶尔也遵守。但在生产环境里这种“偶尔”是不可接受的。1.2 Tool Call把“文本”变成了“参数”Tool Call的核心思路是不要直接让模型生成JSON文本而是让模型“决定调用某个工具”工具的参数定义由开发者严格指定。模型输出的是工具名和参数对象这个参数对象本身就是结构化的。举个例子。传统方式下你让模型输出“投诉单信息”{customer_name: 张三, category: 售后, content: 商品破损}而在Tool Call模式下你先给模型一个工具定义{ name: save_complaint, parameters: { type: object, properties: { customer_name: {type: string}, category: {type: string, enum: [售后, 物流, 商品质量]}, content: {type: string} }, required: [customer_name, category, content] } }模型输出的是一个结构化的“调用意图”而不是一段需要二次解析的文本。框架内部已经帮你把参数JSON解析好了你直接拿到的就是一个Python字典或Pydantic对象。这就是从“让模型写文本”到“让模型填参数”的本质转变。1.3 为什么要用withStructuredOutput再封装一层既然Tool Call已经能解决结构问题那直接用底层调用不就行了吗理论上可以但实际操作起来代码很啰嗦。你需要手动构造工具定义JSON Schema。把Schema塞进请求的tools参数。解析返回的tool_calls字段。做类型校验失败还要重试。这几步写下来一个“简单”调用就要20多行代码。withStructuredOutput的定位就是把这套样板逻辑收进一行调用里让你专注于“结构长什么样”而不是“怎么把结构弄出来”。2. withStructuredOutput 一行封装的核心机制withStructuredOutput在不同框架里都有类似实现比如LangChain的bind_tools加with_structured_output组合拳或者一些云服务SDK里的同名方法核心思路是一致的以“类型定义”为核心自动完成Tool Call的声明、调用和结果校验。2.1 最简单的一行调用假设我们有一个Pydantic模型定义了要抽取的数据结构from pydantic import BaseModel, Field from typing import Literal class Complaint(BaseModel): customer_name: str Field(description客户姓名) category: Literal[售后, 物流, 商品质量] Field(description投诉分类) content: str Field(description投诉内容) order_id: str | None Field(defaultNone, description关联订单号)调用方式变得非常直接structured_llm llm.with_structured_output(Complaint) result structured_llm.invoke(客户张三反馈商品破损要求补发订单号ORD20250111)这行代码执行完成之后result直接就是一个Complaint实例print(result.customer_name) # 张三 print(result.category) # 售后 print(result.content) # 客户反馈商品破损要求补发 print(result.order_id) # ORD20250111整个过程你没有写任何JSON解析逻辑没有处理tool_calls的返回结构也不需要担心模型多输出的废话文本。底层框架已经把“定义Schema、声明工具、解析结果、校验字段”全部做完了。2.2 底层到底做了什么如果你好奇这“一行封装”背后发生了什么拆解下来主要有四步根据你传入的Complaint类自动生成JSON Schema。把这个Schema注入到请求的tools参数里同时设置tool_choice强制模型调用该工具。收到响应后提取tool_calls参数部分。用Pydantic或其他校验器反序列化成目标对象如果校验失败有些框架会触发自动重试。这个“自动重试”是节省开发者时间的杀手级特性。传统手工方案里模型返回一个非法JSON你要自己判断“是重试还是降级处理”然后重新构建prompt。而withStructuredOutput内置的重试机制会把失败信息作为反馈再次提交给模型让模型自己纠正。这就像你把草稿交给模型模型看一眼说格式不对然后重新起草一版直到格式合规。2.3 对比传统方案的代码量我拿同一个需求做个对比。传统手工Tool Call方案长这样tool_schema { name: extract_complaint, description: 提取投诉信息, parameters: Complaint.model_json_schema() } messages [{role: user, content: 客户张三反馈商品破损要求补发}] response llm.invoke(messages, tools[tool_schema], tool_choicerequired) tool_call response.tool_calls[0] arguments json.loads(tool_call[args]) try: result Complaint.model_validate(arguments) except ValidationError as e: logger.warning(f校验失败: {e}触发一次重试) messages.append({role: user, content: f上次输出格式非法: {e}请重新生成}) response llm.invoke(messages, tools[tool_schema], tool_choicerequired) tool_call response.tool_calls[0] arguments json.loads(tool_call[args]) result Complaint.model_validate(arguments)这还没算上处理“模型没有调用工具”的兜底分支。而用withStructuredOutput上面十几行代码直接缩减为一行而且异常处理、重试策略都是经过框架默认设计优化的。对于需要大量做数据抽取、信息分类的工程来说这行封装省下来的可不止是敲键盘时间还有数不清的边界情况处理。2.4 使用withStructuredOutput时的注意点在我实际使用中有几个细节需要特别留意枚举字段尽量用Literal而不是普通str。Literal[售后, 物流]会让模型只能在这几个选项中选极大降低自由发挥的概率。可空字段一定要给defaultNone。否则模型如果没提取到对应信息校验就会报错导致无谓重试。描述信息要写清楚。Pydantic字段的description会出现在Schema里模型的输出质量高度依赖这些描述。比如content字段如果写成“投诉内容”模型的提取完整度不如“用户的原始投诉描述保留关键细节”。不要塞一个巨大的嵌套结构。Schema越复杂模型出错的概率越高。尽量把结构控制在两层以内超过两层的可以拆成多次子调用。3. 流式输出与结构化结果的协作方案聊完了基础封装再来看一个更贴合实际生产场景的问题流式输出。很多应用为了让用户等待时体验更好会选择把模型输出一个字一个字地“流”给前端。这个需求和结构化输出冲突吗表面看有一点因为结构化输出要求的是“完整、严格的JSON”而流式输出要求的是“先让用户看到内容”。3.1 流式输出的本质LLM的流式输出Streaming本质上是让模型像打字一样逐token生成内容。对于普通文本我们可以在每个token到达时立刻推送到前端用户看到的是逐字出现的效果。但对于Tool Call模式模型输出的“内容”其实是工具调用的参数JSON片段这些片段对用户来说没有阅读价值直接推给前端会看到一个JSON字符串在“打字机”式闪烁。所以流式模式下的Tool Call需要区分两种场景场景A模型还在一本正经地生成文本内容给用户看。场景B模型决定调用工具开始输出参数JSON。3.2 withStructuredOutput 在流式下怎么用在不少框架里withStructuredOutput支持通过stream方法获取流式输出。关键点在于流式输出在token级别仍然是“非结构化的”真正的结构化结果会在流结束后由框架汇总成完整的最终对象。structured_llm llm.with_structured_output(Complaint) # 假设有一个用户输入 user_input 客户张三反馈商品破损要求补发 stream structured_llm.stream(user_input) for chunk in stream: # chunk可能包含两种内容 # 1. 普通文本片段比如模型先说了句“正在为你处理” # 2. 结构化的增量信息 print(chunk)这种场景下正确做法是把普通文本片段实时渲染到前端让用户体验到“模型在思考回应”的流畅感。同时把结构化的工具调用增量累积在后台等流结束后再统一处理。最终获得的结构化对象和普通invoke调用是一样的也是Complaint实例。也就是说withStructuredOutput保证了无论走流式还是非流式最后的数据形态保持一致。这就让上层业务代码可以无差别处理两种模式的返回值不用为管道切换写两套逻辑。3.3 实操流式渲染 结构化落库这里给出一段结合流式渲染和结构化收集的完整示例便于读者参考from collections import deque def process_complaint_streaming(user_input: str): structured_llm llm.with_structured_output(Complaint) # 用一个队列暂存结构化片段 chunk_buffer deque() # 第一遍流式实时展示文本内容 for chunk in structured_llm.stream(user_input): # 如果是纯文本内容则打印到前端 if hasattr(chunk, content) and chunk.content: print(f[实时输出]: {chunk.content}, flushTrue) # 如果有结构化增量则放入缓冲区 if getattr(chunk, tool_call_chunks, None): chunk_buffer.append(chunk.tool_call_chunks) # 流式结束后重新用不流式的方式取一次完整结构化结果 # 需要注意有些框架的stream结果会累积出最终对象如果不支持 # 则需要在流结束后用相同的参数再调用一次invoke获取最终结果 final_result structured_llm.invoke(user_input) # 此时final_result为Complaint实例 print(f最终结构化结果: {final_result.model_dump()}) return final_result这里有个工程上的取舍要说明白为了拿到“最终完整对象”有些框架的stream会在内部帮你累积有些框架则要求你流结束后再调用一次invoke。后者的代价是二次请求成本略高。如果对延迟敏感可以检查一下你用的框架是否支持stream返回的最后一个消息里带tool_calls完整结构如果有直接用那个结果就行。我的建议是在容忍一次额外调用的场景下老老实实二次请求逻辑更简单也不容易踩“流式累积不完整”的坑。对于追求极致成本的场景再去做缓存和结构复用。3.4 流式场景的降级策略流式输出在用户延迟体验上占优势但稳定性上并不比普通调用强。如果模型本身不支持工具调用或者网络中断导致流中断你需要有一个降级方案。我的做法是在主链路使用withStructuredOutput。如果模型支持include_rawTrue部分框架会返回原始文本和结构化对象可以在流结束后拿到原始文本做一次兜底解析。如果结构化对象仍然校验失败返回一个“整理失败”的响应等待用户重试。这套策略在真实项目中运行了很长时间用户体验损失很小因为绝大多数情况下结构化调用是能成功的。降级只是兜底不是常态。4. MySQL 落地从模型对象到数据行拿到Complaint对象之后业务上还需要把它存起来。这块的难点不在于SQL语法而在于数据一致性、重复提交处理和历史数据沉淀。4.1 表结构设计结构化输出落地到MySQL首要问题是表怎么建。很多初学者喜欢把模型返回的JSON整个塞进一个TEXT字段一了百了。这样做的缺点是无法针对单个字段建立索引无法在数据库层面做约束后续分析数据时要依赖程序反序列化。更推荐的做法是“主体字段单独建列附带一个JSON字段存扩展信息”。以投诉单为例CREATE TABLE complaint_records ( id BIGINT AUTO_INCREMENT PRIMARY KEY, task_id VARCHAR(64) NOT NULL COMMENT 业务幂等键, customer_name VARCHAR(128) NOT NULL COMMENT 客户姓名, category VARCHAR(32) NOT NULL COMMENT 投诉分类, content TEXT NOT NULL COMMENT 投诉内容, order_id VARCHAR(64) DEFAULT NULL COMMENT 关联订单号, raw_json JSON NOT NULL COMMENT 模型原始结构化输出, model_name VARCHAR(64) DEFAULT NULL COMMENT 使用的模型标识, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_task_id (task_id), KEY idx_category (category), KEY idx_order_id (order_id), KEY idx_created_at (created_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENTLLM投诉信息提取结果表;raw_json字段保存模型返回的完整JSON目的是保留原始证据方便后续审计也方便模型迭代后重新分析。customer_name、category、content这些核心字段单独成列是给查询和分析用的。这种设计平衡了存储的通用性和查询的效率。如果你确实不需要对单个字段做查询或过滤只留一个raw_json也够用但生产环境我推荐至少把主键字段、分类字段、时间字段抽出来。4.2 写入策略与幂等性LLM应用有个天然问题同一段输入你今天跑一遍和明天跑一遍模型输出的内容可能略有差异temperature较高时更明显。如果业务允许你可能需要“重复执行抽取”来对比效果此时就可能产生重复数据。解决办法是引入幂等键。我在设计task_id时就做了双层处理前端或业务层生成一个唯一任务ID比如complaint_20250211_123456每次重试都用同一个ID。数据库里对task_id建唯一索引写入时用INSERT ... ON DUPLICATE KEY UPDATE或者先查后插。写入代码示例使用pymysqlimport json from datetime import datetime def save_complaint_to_mysql(complaint: Complaint, task_id: str, model_name: str): conn get_mysql_connection() try: with conn.cursor() as cursor: sql INSERT INTO complaint_records (task_id, customer_name, category, content, order_id, raw_json, model_name) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE customer_nameVALUES(customer_name), categoryVALUES(category), contentVALUES(content), order_idVALUES(order_id), raw_jsonVALUES(raw_json), model_nameVALUES(model_name) cursor.execute(sql, ( task_id, complaint.customer_name, complaint.category, complaint.content, complaint.order_id, json.dumps(complaint.model_dump(), ensure_asciiFalse), model_name )) conn.commit() finally: conn.close()这个方案的好处是“跑得多但不会重复建单”每次重试只会覆盖同一条任务ID对应的数据。对于LLM应用这非常重要因为你无法保证模型第一次调用就能给出完美结果重试是常态。4.3 批量写入与性能优化如果是离线大批量抽取比如你有一百万条用户反馈文本要交给LLM处理那不能一条一条地插入数据库。至少要做以下优化批量拼接SQL执行插入比如每100条一次executemany。使用连接池避免每次创建新连接。关闭自动提交全部处理完再统一commit减少磁盘同步频率。示例代码def batch_save_to_mysql(complaints: list[Complaint], task_ids: list[str], model_name: str): conn get_connection_from_pool() batch_size 100 try: with conn.cursor() as cursor: for i in range(0, len(complaints), batch_size): batch_data [] for j in range(i, min(i batch_size, len(complaints))): c complaints[j] batch_data.append(( task_ids[j], c.customer_name, c.category, c.content, c.order_id, json.dumps(c.model_dump(), ensure_asciiFalse), model_name )) cursor.executemany(insert_sql_with_upsert, batch_data) conn.commit() finally: conn.close()注意MySQL插入时TEXT和JSON字段都有单条大小限制max_allowed_packet如果模型输出的content特别长比如上万字可能触发“Packet too large”异常。遇到这个情况要么调大MySQL的max_allowed_packet参数要么在写入前做截断补偿——我的做法是先尝试完整写入失败后把content截断到3000字并记录一条警告日志确保数据不丢失。5. 常见问题与排查技巧实录从withStructuredOutput调用到MySQL落地链路不算长但每一段都有各自的坑。我把实际运行中遇到的高频问题整理成速查表方便大家排雷。问题阶段现象常见原因解决建议结构化调用返回对象字段全部为默认值prompt信息不足模型未提取到关键信息优化字段description调整prompt引导结构化调用枚举字段校验失败模型返回了非枚举值如“退换”不在列表中枚举值补充到prompt中增加自动重试次数结构化调用复杂嵌套结构频繁失败Schema带头太深模型生成错误拆分成多次简单调用二次合并流式处理流式返回的对象不完整框架版本差异流没有累积tool_calls流结束后用invoke补充获取MySQL写入Packet too largecontent字段内容过长调大max_allowed_packet或截断MySQL写入重复任务ID覆盖历史数据upsert逻辑覆盖了旧记录确认幂等键设计是否符合业务语义MySQL写入JSON字段存入乱码连接字符集未设utf8mb4连接参数加charsetutf8mb45.1 模型不肯“填表”怎么办withStructuredOutput通过Tool Call强制模型输出结构化参数但个别小模型可能不理解工具调用协议导致怎么调都返回普通文本。遇到这种情况检查几件事你使用的模型是否明确支持Tool Call工具调用能力。确认没有在请求里误关tool_choice。尝试给模型发送一条示例在prompt里加入“请使用提供的工具处理”这样的引导。如果以上都不行换一个支持工具调用的模型是最直接的解决办法。工具调用能力不是所有模型都有的标配这是个硬门槛。5.2 温度参数对结构化结果的影响你可能在调LLM时习惯把temperature设成0.7甚至更高以获得更“有创意”的回答。但结构化抽取场景我会把temperature设为0完全确定性。原因很简单抽取信息越稳定越好。模型的“创意”在抽取场景里毫无意义而temperature0可以最大程度保证同一段输入在多次调用中得到一致的结果这在数据生产线上很关键。withStructuredOutput一般不会帮你改温度参数你得在构造LLM时自己设置llm SomeLLM(temperature0)5.3 结构化输出后的数据审计最后说一个容易被忽略的点。把模型结果直接写入业务库之前最好走一层人工复核或者规则校验。我在项目里加了几个简单规则customer_name长度至少2个字符。category必须在预定枚举范围内。content不为空且长度小于5000。规则校验不通过的数据会进入“待人工处理”池而不是自动写入业务表。这样设计让系统有了“刹车”能力避免模型偶尔的输出异常污染核心数据。这些规则不需要复杂引擎一个简单的Python函数就能搞定。写在最后从流式输出到Tool Call再到withStructuredOutput一行封装最后落到MySQL整条链路在工程上的核心就一句话让模型少自由发挥让数据多一层约束。我在实际项目中最大的体会是LLM的结构化输出不是靠“祈祷模型乖乖听话”来实现的而是靠框架机制把模型的自由度锁死在Schema定义范围内。withStructuredOutput的出现确实让这一步变得干净利落。最后分享一个小技巧如果你的团队后期需要频繁改字段结构建议把Pydantic模型定义抽成一个独立的schemas.py文件并且给模型加一个version字段。这样即使以后Schema升级了历史数据仍然能追溯到当初是用哪个版本的结构抽取出来的排查线上问题时会省很多力气。