这次我们来看一个在AI开发中非常实际的问题：如何让大模型稳定、可靠地输出结构化的JSON数据。无论是构建AI Agent、开发自动化工具，还是处理复杂的API调用，JSON格式的稳定输出都是连接大模型能力与下游业务逻辑的关键桥梁。然而，开发者常常遇到模型输出格式飘忽不定、JSON解析失败、或内容不符合预定Schema的困扰。

这篇文章不讨论抽象概念，直接聚焦于可落地的工程化解决方案。我们将拆解从提示词设计、调用策略到后处理校验的全链路方法，让你能快速在自己的项目中应用，确保大模型返回的JSON数据既“能用”又“好用”。

如果你正在开发基于大模型的Agent、需要处理结构化数据抽取，或者正在为“大模型面试”中相关的工程问题寻找答案，那么接下来的内容将直接提供可复用的代码和清晰的排查思路。

1. 核心能力速览：构建稳定JSON输出的工具箱

在深入细节之前，我们先快速梳理一下确保大模型稳定输出JSON所涉及的核心技术环节和工具。这能帮助你快速判断哪些方法适合你的场景。

2. 适用场景与使用边界

2.1 谁需要关注JSON的稳定输出？

• AI Agent开发者：Agent的每一步决策（工具调用、状态判断）都需要结构化的输出。
• 后端工程师：需要将大模型的自然语言能力集成到现有系统中，返回的数据必须能被程序直接消费。
• 数据分析师/产品经理：希望通过自然语言查询，自动生成结构化的数据报告或看板。
• 面试官/学习者：在“大模型面试”中，如何让模型按指定格式输出答案，本身就是考察工程化思维和提示词工程能力的经典题目。

2.2 能解决什么问题？

1. 格式一致性：避免模型时而返回JSON，时而返回一段包含JSON的文本，甚至纯自然语言。
2. 字段完整性：确保返回的JSON包含所有必需的字段，不缺不漏。
3. 类型安全：确保字段的值类型（字符串、数字、数组、布尔值）符合预期，避免后续处理出错。
4. 提高系统鲁棒性：通过自动化的校验和修复机制，降低人工干预成本，提升整个AI工作流的可靠性。

2.3 不适合什么场景？

• 完全自由的创意写作：如果需要模型天马行空地创作，强制JSON输出会限制其发挥。
• 极其简单的一次性任务：如果只是偶尔让模型总结一段话，直接处理纯文本可能更简单。
• 对延迟极其敏感的场景：复杂的多轮校验和重试机制必然会增加响应时间。

2.4 合规与安全边界

• 数据隐私：通过大模型API处理数据时，需遵守相关服务条款，避免传输敏感个人信息。
• 内容审核：对于生成的内容，应有后续审核机制，确保符合法律法规和公序良俗，即使它被包装在JSON中。
• 授权使用：确保用于微调或提供上下文的数据拥有合法授权。

3. 环境准备与前置条件

在开始编码之前，你需要准备好基础环境。以下是一个通用清单，请根据你选择的具体模型和框架进行调整。

1. Python环境：推荐使用 Python 3.8 及以上版本。使用conda或venv创建独立的虚拟环境是最佳实践。

   # 创建并激活虚拟环境 (以conda为例) conda create -n structured_llm python=3.10 conda activate structured_llm

2. 大模型访问权限：

   • 云端API：获取OpenAI、Anthropic (Claude)、DeepSeek、智谱AI等任一服务的API Key。
   • 本地模型：如果你使用Ollama、vLLM、LM Studio等工具本地部署模型（如Qwen、Llama），确保模型服务已启动并可访问。

3. 核心Python库：安装以下常用库。

   pip install openai anthropic requests pydantic langchain langchain-openai jsonschema

   • openai/anthropic：官方SDK。
   • pydantic：用于定义数据模型和验证，是LangChain Output Parser的基石。
   • langchain：提供丰富的Output Parser和Chain组件，能大幅简化开发。
   • jsonschema：用于更灵活的JSON Schema校验。

4. 代码编辑器或IDE：如VS Code、PyCharm等。

4. 方法一：强化系统提示词（System Prompt）

这是最直接、成本最低的方法，适用于所有支持系统提示词的大模型。

4.1 基础指令模板

在你的系统提示词中，必须清晰、强硬地指定输出格式。不要用“请”、“最好”这类模糊词汇。

一个效果较差的示例：

请你分析用户情绪，并返回一个JSON对象。

一个强约束的示例：

你是一个情绪分析助手。你必须严格按照以下JSON格式输出，不要输出任何其他解释、前缀或后缀。 输出格式： { "emotion": "字符串，必须是‘positive‘, ‘negative‘, ‘neutral‘中的一个", "confidence": "浮点数，范围0.0到1.0", "keywords": "字符串数组，列出支撑该情绪判断的关键词" } 现在，开始分析。

4.2 实战代码示例

假设我们使用OpenAI API。

import openai import json client = openai.OpenAI(api_key="your-api-key") def analyze_emotion_with_prompt(text): system_prompt = """你是一个情绪分析助手。你必须严格按照以下JSON格式输出，不要输出任何其他解释、前缀或后缀。 输出格式： { "emotion": "字符串，必须是‘positive‘, ‘negative‘, ‘neutral‘中的一个", "confidence": "浮点数，范围0.0到1.0", "keywords": "字符串数组，列出支撑该情绪判断的关键词" } """ try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"分析以下文本的情绪：{text}"} ], temperature=0.1, # 降低随机性，使输出更稳定 max_tokens=150 ) raw_output = response.choices[0].message.content.strip() # 尝试直接解析 result = json.loads(raw_output) return result except json.JSONDecodeError as e: print(f"JSON解析失败！原始输出：{raw_output}") # 此处可加入后处理逻辑，见方法四 return None # 测试 test_text = “这个产品真是太棒了，用户体验完美，我会推荐给朋友！” result = analyze_emotion_with_prompt(test_text) if result: print(f"解析成功：{result}")

效果验证：

• 成功标准：函数返回一个合法的Python字典，且包含emotion,confidence,keywords三个键。
• 失败排查：检查raw_output，看模型是否添加了额外的Markdown代码块标记（如json ...），或者是否输出了解释性文字。这需要在后处理步骤中处理。

5. 方法二：使用原生函数调用（Function Calling）

OpenAI、DeepSeek等API提供了函数调用功能，这是目前最稳定、最官方的结构化输出方式。模型会输出一个符合预定参数的函数调用请求，而非直接的JSON字符串。

5.1 定义“工具”（函数）

你需要将你期望的输出结构，定义为一个虚拟的“函数”。

import openai import json client = openai.OpenAI(api_key="your-api-key") def analyze_emotion_with_function_calling(text): tools = [ { "type": "function", "function": { "name": "record_emotion_analysis", "description": "记录对一段文本的情绪分析结果", "parameters": { "type": "object", "properties": { "emotion": { "type": "string", "enum": ["positive", "negative", "neutral"], "description": "情绪分类" }, "confidence": { "type": "number", "description": "置信度，0.0到1.0之间" }, "keywords": { "type": "array", "items": {"type": "string"}, "description": "关键词列表" } }, "required": ["emotion", "confidence", "keywords"], "additionalProperties": False # 禁止额外字段，保证结构纯净 } } } ] try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "user", "content": f"分析以下文本的情绪：{text}"} ], tools=tools, tool_choice={"type": "function", "function": {"name": "record_emotion_analysis"}}, # 强制调用特定函数 temperature=0.1 ) # 提取函数调用参数 tool_call = response.choices[0].message.tool_calls[0] if tool_call.function.name == "record_emotion_analysis": arguments_str = tool_call.function.arguments result = json.loads(arguments_str) # 这里直接就是标准JSON return result else: return None except (AttributeError, IndexError, json.JSONDecodeError) as e: print(f"函数调用解析失败：{e}") return None # 测试 test_text = “这次更新后，软件频繁崩溃，让我非常失望。” result = analyze_emotion_with_function_calling(test_text) if result: print(f"通过函数调用解析成功：{result}") print(f"情绪：{result[‘emotion‘]}, 置信度：{result[‘confidence‘]}")

核心优势：

• 极高稳定性：API设计保证了输出严格遵循你定义的JSON Schema。
• 类型安全：参数类型（string, number, array等）被严格约束。
• 无需后处理：直接获得解析好的字典。

6. 方法三：利用LangChain的Pydantic Output Parser

如果你在使用LangChain框架，PydanticOutputParser是集成度和便捷性最高的选择。它结合了Pydantic的数据验证和LangChain的提示词模板。

6.1 定义输出数据结构

首先，用Pydantic定义一个数据模型。

from pydantic import BaseModel, Field from typing import List from langchain.output_parsers import PydanticOutputParser from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI # 1. 定义你的数据结构 class EmotionAnalysis(BaseModel): emotion: str = Field(description="情绪分类", enum=["positive", "negative", "neutral"]) confidence: float = Field(description="置信度，0.0到1.0", ge=0.0, le=1.0) keywords: List[str] = Field(description="关键词列表") # 2. 创建解析器 parser = PydanticOutputParser(pydantic_object=EmotionAnalysis) # 3. 创建提示词模板，自动注入格式指令 prompt_template = """ 你是一个情绪分析助手。 {format_instructions} 请分析以下文本的情绪： {user_input} """ prompt = PromptTemplate( template=prompt_template, input_variables=["user_input"], partial_variables={"format_instructions": parser.get_format_instructions()} # 关键！自动生成格式说明 ) # 4. 构建Chain model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1, openai_api_key="your-api-key") chain = prompt | model | parser # 使用LangChain表达式语法(LCEL) # 5. 调用 def analyze_with_langchain(text): try: result = chain.invoke({"user_input": text}) return result except Exception as e: print(f"LangChain解析失败：{e}") # 可以在这里获取原始输出进行修复 # raw_output = ... return None # 测试 test_text = “天气不错，心情很好。” result = analyze_with_langchain(test_text) if result: print(f"LangChain Pydantic解析成功：{result}") print(type(result)) # <class ‘__main__.EmotionAnalysis‘> # 可以像对象一样访问属性 print(f"情绪对象：{result.emotion}, 置信度：{result.confidence}")

parser.get_format_instructions()生成的指令示例：

The output should be formatted as a JSON instance that conforms to the JSON schema below. As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]} the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted. Here is the output schema:

{"properties": {"emotion": {"title": "Emotion", "description": "情绪分类", "enum": ["positive", "negative", "neutral"], "type": "string"}, "confidence": {"title": "Confidence", "description": "置信度，0.0到1.0", "maximum": 1.0, "minimum": 0.0, "type": "number"}, "keywords": {"title": "Keywords", "description": "关键词列表", "type": "array", "items": {"type": "string"}}}, "required": ["emotion", "confidence", "keywords"]}

这种方法将格式约束作为系统的一部分，非常优雅，且能自动处理很多边界情况。 ## 7. 方法四：后处理、校验与重试机制 无论前几种方法多么完善，网络波动、模型抽风都可能产生意外输出。一个健壮的系统必须包含后处理层。 ### 7.1 防御性JSON解析 ```python import json import re def robust_json_parse(raw_text: str): """ 尝试从可能包含额外字符的文本中提取并解析JSON。 """ if not raw_text: return None # 情况1：输出被包裹在 ```json ... ``` 标记中 json_code_block = re.search(r‘```(?:json)?\s*([\s\S]*?)\s*```‘, raw_text) if json_code_block: raw_text = json_code_block.group(1).strip() # 情况2：输出是纯JSON，但可能有首尾空白 raw_text = raw_text.strip() # 尝试直接解析 try: return json.loads(raw_text) except json.JSONDecodeError: pass # 情况3：输出可能包含“输出为：{...}”或“JSON: {...}”等前缀 # 尝试寻找第一个‘{‘和最后一个‘}‘ start_idx = raw_text.find(‘{‘) end_idx = raw_text.rfind(‘}‘) if start_idx != -1 and end_idx != -1 and start_idx < end_idx: json_str = raw_text[start_idx:end_idx+1] try: return json.loads(json_str) except json.JSONDecodeError: pass # 情况4：作为最后手段，尝试用`ast.literal_eval`评估类似Python字典的字符串（谨慎使用） # 这里略过，因为安全性需要考虑。 print(f“无法从文本中解析JSON：{raw_text[:200]}...“) return None

7.2 基于JSON Schema的校验

即使解析成功，内容也可能不符合要求。使用jsonschema库进行验证。

import jsonschema from jsonschema import validate # 定义Schema emotion_schema = { "type": "object", "properties": { "emotion": {"type": "string", "enum": ["positive", "negative", "neutral"]}, "confidence": {"type": "number", "minimum": 0, "maximum": 1}, "keywords": {"type": "array", "items": {"type": "string"}} }, "required": ["emotion", "confidence", "keywords"], "additionalProperties": False } def validate_with_schema(data): try: validate(instance=data, schema=emotion_schema) return True, “校验通过“ except jsonschema.ValidationError as e: return False, f“Schema校验失败：{e.message}“ # 使用示例 parsed_data = {“emotion“: “happy“, “confidence“: 0.9, “keywords“: [“good“]} # “happy“不在enum中 is_valid, msg = validate_with_schema(parsed_data) print(f“校验结果：{is_valid}, 信息：{msg}“)

7.3 自动重试机制

将解析、校验和重试封装起来。

def get_structured_output_with_retry(prompt_func, user_input, max_retries=3): """ prompt_func: 一个函数，接收user_input，返回大模型的原始输出字符串。 """ for attempt in range(max_retries): print(f“第 {attempt + 1} 次尝试...“) raw_output = prompt_func(user_input) parsed_data = robust_json_parse(raw_output) if parsed_data is None: print(“JSON解析失败，准备重试。“) continue is_valid, error_msg = validate_with_schema(parsed_data) if is_valid: return parsed_data else: print(f“Schema校验失败：{error_msg}，准备重试。“) print(f“经过 {max_retries} 次重试后仍失败。“) return None # 模拟一个会随机出错的提示词函数 def unreliable_llm_call(text): import random responses = [ ‘{“emotion“: “positive“, “confidence“: 0.8, “keywords“: [“great“, “awesome“]}‘, ‘Here is the result: {“emotion“: “neutral“, “confidence“: 0.5, “keywords“: [“okay“]}‘, ‘```json\n{“emotion“: “negative“, “confidence“: 0.9, “keywords“: [“bad“, “terrible“]}\n```‘, ‘I think the emotion is positive.‘, # 这个会失败 ‘{“feeling“: “good“, “score“: 0.7}‘ # 这个Schema不对 ] return random.choice(responses) result = get_structured_output_with_retry(unreliable_llm_call, “test“, max_retries=5) print(f“最终结果：{result}“)

8. 接口API与批量任务设计

当你拥有一个稳定的JSON生成核心后，可以将其封装成服务，并支持批量处理。

8.1 构建FastAPI服务

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import asyncio from your_llm_module import get_structured_output_with_retry # 导入前面封装好的函数 app = FastAPI(title=“大模型结构化输出API“) class AnalysisRequest(BaseModel): text: str max_retries: int = 3 class AnalysisResponse(BaseModel): success: bool data: dict = None error: str = None @app.post(“/analyze/emotion“, response_model=AnalysisResponse) async def analyze_emotion(request: AnalysisRequest): “““单个文本情绪分析接口””” result = get_structured_output_with_retry(your_prompt_function, request.text, request.max_retries) if result: return AnalysisResponse(success=True, data=result) else: return AnalysisResponse(success=False, error=“分析失败，请重试或检查输入。“) class BatchAnalysisRequest(BaseModel): tasks: List[AnalysisRequest] class BatchAnalysisResponse(BaseModel): results: List[AnalysisResponse] @app.post(“/analyze/emotion/batch“, response_model=BatchAnalysisResponse) async def analyze_emotion_batch(request: BatchAnalysisRequest): “““批量文本情绪分析接口””” tasks = [] for task in request.tasks: # 为每个任务创建异步协程 tasks.append(asyncio.create_task( process_single_analysis(task.text, task.max_retries) )) # 并发执行 results = await asyncio.gather(*tasks, return_exceptions=True) formatted_results = [] for r in results: if isinstance(r, Exception): formatted_results.append(AnalysisResponse(success=False, error=str(r))) else: formatted_results.append(r) return BatchAnalysisResponse(results=formatted_results) async def process_single_analysis(text: str, max_retries: int) -> AnalysisResponse: # 这里模拟一个异步处理函数 # 在实际应用中，这里应该调用异步的LLM客户端 try: # 假设 get_structured_output_with_retry 是同步的，可以放在线程池中运行以避免阻塞事件循环 import concurrent.futures loop = asyncio.get_event_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor( pool, get_structured_output_with_retry, your_prompt_function, text, max_retries ) if result: return AnalysisResponse(success=True, data=result) else: return AnalysisResponse(success=False, error=“分析失败“) except Exception as e: return AnalysisResponse(success=False, error=f“处理异常：{e}“) # 启动命令：uvicorn api:app --reload --host 0.0.0.0 --port 8000

8.2 调用示例

启动服务后，可以使用curl或Python客户端调用。

# 单条请求 curl -X POST “http://127.0.0.1:8000/analyze/emotion“ \ -H “Content-Type: application/json“ \ -d ‘{“text“: “今天真是糟糕的一天。“}‘ # 批量请求 curl -X POST “http://127.0.0.1:8000/analyze/emotion/batch“ \ -H “Content-Type: application/json“ \ -d ‘{ “tasks“: [ {“text“: “产品很棒！“}, {“text“: “服务一般。“}, {“text“: “非常失望！“} ] }‘

# Python客户端调用示例 import requests import json url = “http://127.0.0.1:8000/analyze/emotion/batch“ payload = { “tasks“: [ {“text“: “产品很棒！“, “max_retries“: 2}, {“text“: “服务一般。“}, {“text“: “非常失望！“} ] } headers = {‘Content-Type‘: ‘application/json‘} response = requests.post(url, data=json.dumps(payload), headers=headers) print(response.json())

9. 性能观察与最佳实践

9.1 性能与资源考量

• 延迟：函数调用和Output Parser通常比纯提示词方法慢几十到几百毫秒，因为涉及更多的序列化/反序列化。重试机制会显著增加延迟（重试次数 * 单次请求时间）。
• Token消耗：详细的格式说明和Schema会占用一部分提示词Token，增加成本。
• 可靠性提升：牺牲少量性能和成本，换取近乎100%的结构化输出成功率，对于生产系统通常是值得的。

9.2 最佳实践清单

1. 从简单开始：先尝试强约束的系统提示词，如果满足要求（如95%+成功率），则无需引入更复杂的方案。
2. 优先使用函数调用：如果使用的API支持（如OpenAI, DeepSeek），这是最稳定、最推荐的方式。
3. 善用LangChain：如果你的项目已经在使用LangChain，PydanticOutputParser能提供非常好的开发体验。
4. 必须实现后处理与重试：无论前段方法多可靠，都要有最后一道防线。robust_json_parse+JSON Schema校验是黄金组合。
5. 设置合理的重试次数与退避：对于非关键任务，2-3次重试足够。对于关键任务，可以考虑指数退避重试，并加入人工审核队列。
6. 监控与日志：记录每次调用的原始输出、解析结果、重试次数和最终状态。这有助于优化提示词和发现模型边界。
7. 分离逻辑与提示：将JSON Schema定义、提示词模板、解析逻辑放在配置文件中，便于维护和A/B测试。
8. 测试覆盖：编写单元测试，模拟模型各种“奇怪”的输出（如包含Markdown、前缀文本、残缺JSON等），确保你的后处理管道能正确处理。

10. 常见问题与排查方法

11. 总结与下一步

让大模型稳定输出JSON不是一个“玄学”问题，而是一个可以通过工程化手段系统性解决的挑战。核心思路是“约束（提示词/函数调用）+ 验证（解析器/Schema）+ 容错（重试/后处理）”的三层保障。

对于大多数应用场景，建议的实践路径是：

1. 第一步：采用“强系统提示词 + 防御性JSON解析”组合，快速验证可行性。
2. 第二步：如果稳定性不达标，优先切换到模型原生的函数调用（Function Calling）方案。
3. 第三步：如果使用LangChain生态，用PydanticOutputParser来获得更优雅的开发体验。
4. 第四步：无论哪种方法，都必须实现基于JSON Schema的校验和有限次数的重试机制，这是生产系统的安全网。

下一步，你可以：

• 将这套流程封装成公司内部的通用SDK或中间件。
• 探索对本地模型进行微调，专门强化其输出指定JSON格式的能力。
• 结合向量数据库，实现更复杂的、基于结构化输出的Agent决策流程。
• 针对“大模型面试”场景，设计一套自动评估JSON输出准确性和合规性的评测系统。

希望这篇从实战出发的梳理，能帮助你彻底解决大模型输出JSON的稳定性问题，让你的AI应用更加可靠和强大。