在构建基于大模型的智能应用时,你是否遇到过这样的困扰:你向模型提问“列出三个用户信息,包括姓名、年龄和邮箱”,期望得到一个结构化的JSON数组,但模型却返回了一段自由文本,甚至夹杂着Markdown代码块标记?这种输出格式的不稳定性,是AI应用开发,特别是Agent(智能体)开发中一个高频痛点。它不仅增加了后端数据解析的复杂度,更可能导致整个业务流程中断。
本文将深入探讨大模型稳定输出JSON格式的实战方案。无论你是正在开发一个需要精准数据提取的AI Agent,还是在准备相关技术面试,本文都将为你提供从核心原理到工程落地的完整指南。我们将从问题根源出发,逐步拆解Prompt工程、函数调用、后处理校验以及使用专业框架等多种解决方案,并附上可运行的代码示例和避坑清单,帮助你彻底解决JSON输出“抽风”的问题。
1. 为什么大模型输出JSON不稳定?
在深入解决方案之前,我们首先要理解问题的根源。大语言模型(LLM)本质上是基于概率生成文本的序列预测模型,其训练目标是生成“合理”的下一个词元(Token),而非严格遵守特定的数据格式规范。
1.1 不稳定性表现与根源分析
常见的不稳定输出形式:
- 自由文本混杂:模型在JSON前后添加解释性文字,如“好的,这是你要的数据:”或“结果如下:”。
- 格式错误:缺少引号、括号不匹配、键名未加双引号(JSON标准要求必须为双引号)、尾部多余逗号。
- Markdown代码块:返回
json {...},将JSON包裹在Markdown标记中。 - 结构漂移:要求的字段缺失、多出未要求的字段,或数组元素结构不一致。
- 类型错误:数字被输出为字符串(如
"age": “25”),布尔值被写为单词。
根本原因可以归结为三点:
- 训练数据偏差:模型在训练时接触了大量非纯JSON的文本,如技术博客、问答对,其中JSON常被嵌入解释性上下文中。
- Prompt歧义:用户的指令(Prompt)不够精确,模型无法区分你是要“生成JSON”还是“描述JSON”。
- 采样随机性:即使使用低温度(Temperature)设置,模型的生成过程仍有一定随机性,可能导致格式上的微小差异。
1.2 稳定JSON输出的核心需求场景
在以下场景中,格式稳定的JSON输出至关重要:
- AI Agent / 智能体:Agent根据观察和思考,需要调用工具(Tool/Function)。工具调用的参数必须以结构化数据(通常是JSON)的形式精确传递。
- 数据提取与结构化:从非结构化文本(如新闻、报告)中提取实体、关系并转化为数据库可接收的格式。
- API接口集成:大模型作为后端服务,需要向前端或其他服务返回可直接解析的数据对象。
- 自动化工作流:将大模型的输出作为下一个自动化节点的输入,格式错误会导致流程失败。
2. 环境与工具准备
在开始实战前,我们需要搭建一个统一的实验环境。本文将以 OpenAI GPT 系列模型(兼容 OpenAI API 的模型)为例,使用 Python 语言进行演示。其他模型(如 Claude、国产大模型)和语言(如 JavaScript)的思路相通。
2.1 基础环境配置
确保你的开发环境已安装 Python 3.8+。我们主要使用openai和pydantic这两个核心库。
# 创建并进入项目目录 mkdir stable-json-output && cd stable-json-output # 创建虚拟环境(推荐) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装核心依赖 pip install openai pydantic # 可选:安装用于更丰富功能(如框架)的库 # pip install instructor # 用于结构化输出的强大框架 # pip install litellm # 统一多模型调用2.2 获取并配置API密钥
你需要一个 OpenAI API 密钥或兼容 OpenAI API 的服务(如 Azure OpenAI, 国内大模型平台)的密钥。
# config.py 或直接在代码中设置环境变量 import os # 方法一:直接设置(不推荐用于生产环境) # openai.api_key = “your-api-key-here” # 方法二:使用环境变量(推荐) # 在终端中执行:export OPENAI_API_KEY=‘your-api-key-here’ # 或在代码中通过os.environ设置 os.environ[“OPENAI_API_KEY”] = “your-api-key-here” # 如果你使用其他兼容服务,可能还需要设置 base_url # os.environ[“OPENAI_API_BASE”] = “https://api.xxx.com/v1”3. 方案一:精炼Prompt工程法
这是最基础、最直接的方法,通过精心设计提示词来引导模型。关键在于明确性、强制性,并提供高质量示例。
3.1 基础指令强化
一个糟糕的Prompt:“给我一些用户数据。” 一个较好的Prompt:“生成一个包含三个用户对象的JSON数组,每个对象有name(字符串)、age(整数)、email(字符串)字段。只输出JSON,不要有任何其他文字。”
# basic_prompt.py import openai import json def get_completion_basic(prompt): client = openai.OpenAI() response = client.chat.completions.create( model=“gpt-3.5-turbo”, # 或 “gpt-4”, “gpt-4-turbo-preview” messages=[{“role”: “user”, “content”: prompt}], temperature=0.1, # 降低随机性,对格式化输出有益 max_tokens=500 ) return response.choices[0].message.content # 测试基础Prompt prompt = “”” 请生成一个包含三个用户信息的JSON数组。 每个用户是一个对象,包含以下字段: - name: 字符串,表示用户名 - age: 整数,表示年龄 - email: 字符串,表示电子邮箱 请确保输出是**纯粹的、有效的JSON字符串**,不要包含任何Markdown代码块标记(如```json),也不要输出任何解释性文字。 “”” result = get_completion_basic(prompt) print(“原始输出:”) print(result) print(“\n尝试解析:”) try: parsed = json.loads(result) print(“解析成功!”, json.dumps(parsed, indent=2, ensure_ascii=False)) except json.JSONDecodeError as e: print(f“解析失败!错误:{e}”) # 尝试清理常见的非JSON前缀/后缀 cleaned = result.strip() if cleaned.startswith(‘```json’): cleaned = cleaned[7:] elif cleaned.startswith(‘```’): cleaned = cleaned[3:] if cleaned.endswith(‘```’): cleaned = cleaned[:-3] cleaned = cleaned.strip() print(“清理后内容:”, cleaned) try: parsed = json.loads(cleaned) print(“清理后解析成功!”, json.dumps(parsed, indent=2, ensure_ascii=False)) except: print(“清理后仍然无法解析。”)关键点分析:
- 明确结构:清晰定义JSON的根(数组)、元素(对象)和每个字段的名称、类型。
- 强制指令:使用“纯粹的、有效的JSON字符串”、“不要包含任何...”、“只输出JSON”等强约束性语句。
- 降低Temperature:
temperature=0.1使输出更确定,更倾向于遵循指令。
3.2 少样本学习(Few-Shot Learning)
在Prompt中提供输入输出的示例,是引导模型格式最有效的方法之一。
# few_shot_prompt.py import openai import json def get_completion_few_shot(prompt): client = openai.OpenAI() response = client.chat.completions.create( model=“gpt-3.5-turbo”, messages=[{“role”: “user”, “content”: prompt}], temperature=0.1, response_format={ “type”: “json_object” } # 注意:此参数要求模型必须输出JSON对象,对数组有限制 ) return response.choices[0].message.content # 少样本Prompt:包含一个清晰的示例 few_shot_prompt = “”” 你的任务是将用户的自然语言请求转换为一个特定的JSON格式。 例如: 用户请求:“列出两个产品,需要产品名和价格。” 输出必须是以下格式的纯JSON,无任何额外文本: { “products”: [ {“name”: “产品A”, “price”: 100}, {“name”: “产品B”, “price”: 200} ] } 现在,请处理新的请求: 用户请求:“创建三个图书条目,包含标题、作者和出版年份。” 请根据上述示例的格式,输出对应的JSON。 “”” result = get_completion_few_shot(few_shot_prompt) print(“少样本学习输出:”) print(result) try: parsed = json.loads(result) print(“解析成功:”, json.dumps(parsed, indent=2, ensure_ascii=False)) except json.JSONDecodeError as e: print(f“解析错误:{e}”)注意:OpenAI API 提供了response_format={ “type”: “json_object” }参数,它能强制模型输出一个有效的JSON对象。这是一个非常强大的特性,但它有两个重要限制:
- 你必须同时在系统或用户消息中明确要求模型输出JSON。
- 它保证输出是JSON对象(以
{开头),但不保证是JSON数组(以[开头)。如果你需要数组,可能仍需结合Prompt。
4. 方案二:函数调用(Function Calling)与工具使用
这是目前生产环境中实现结构化输出最稳定、最受推荐的方法。模型不直接输出JSON,而是输出一个“意图调用某个函数”的请求,其中参数是结构化的JSON。开发者预先定义好函数(工具)的Schema,模型会严格遵循这个Schema来生成参数。
4.1 使用OpenAI原生函数调用
# function_calling.py import openai import json # 1. 定义我们期望模型能够调用的“函数”(工具)的Schema tools = [ { “type”: “function”, “function”: { “name”: “extract_user_info”, “description”: “从描述中提取用户信息并生成结构化的列表”, “parameters”: { “type”: “object”, “properties”: { “users”: { “type”: “array”, “description”: “用户对象列表”, “items”: { “type”: “object”, “properties”: { “name”: {“type”: “string”, “description”: “用户姓名”}, “age”: {“type”: “integer”, “description”: “用户年龄”}, “email”: {“type”: “string”, “description”: “用户邮箱”} }, “required”: [“name”, “age”, “email”] } } }, “required”: [“users”] } } } ] # 2. 准备用户请求 messages = [{“role”: “user”, “content”: “帮我提取这三个人信息:张三,25岁,zhangsan@example.com;李四,30岁,lisi@example.com;王五,28岁,wangwu@example.com。”}] client = openai.OpenAI() # 3. 发起聊天补全请求,并告知模型可用的工具 response = client.chat.completions.create( model=“gpt-3.5-turbo”, messages=messages, tools=tools, tool_choice=“auto”, # 让模型决定是否调用函数。设为 {“type”: “function”, “function”: {“name”: “extract_user_info”}} 可强制调用 temperature=0 ) # 4. 解析模型的响应 response_message = response.choices[0].message print(“模型原始响应消息:”, response_message) # 5. 检查模型是否决定调用函数 if response_message.tool_calls: # 通常只有一个工具调用 tool_call = response_message.tool_calls[0] if tool_call.function.name == “extract_user_info”: # 提取函数调用参数(这已经是标准的JSON字符串) function_args_str = tool_call.function.arguments print(“\n模型生成的函数参数(JSON字符串):”) print(function_args_str) # 解析JSON try: function_args = json.loads(function_args_str) print(“\n解析后的结构化数据:”) print(json.dumps(function_args, indent=2, ensure_ascii=False)) # 在实际应用中,这里你会调用真实的 extract_user_info 函数 # result = extract_user_info(**function_args) except json.JSONDecodeError as e: print(f“解析函数参数失败:{e}”) else: print(“模型没有选择调用函数,返回了普通文本:”, response_message.content)方案优势:
- 极高稳定性:模型输出的
arguments严格遵循你定义的 JSON Schema,格式错误率极低。 - 类型安全:Schema中定义了类型(string, integer等),模型会尽力遵守。
- 意图明确:将“生成数据”的任务转化为“调用函数”,更符合模型在工具使用场景下的训练目标。
这是构建可靠AI Agent的基石。
5. 方案三:使用专业框架(Instructor)
对于复杂的数据结构,手动编写Prompt和解析逻辑依然繁琐。Instructor库应运而生,它利用Pydantic模型来定义数据结构,并通过修补(patch)OpenAI客户端,将结构化输出的过程极大简化。
5.1 安装与基础使用
首先安装库:pip install instructor
# instructor_basic.py import instructor from openai import OpenAI from pydantic import BaseModel, Field from typing import List # 1. 使用instructor修补OpenAI客户端 client = instructor.patch(OpenAI()) # 2. 使用Pydantic定义你期望的数据结构 class User(BaseModel): name: str = Field(…, description=“用户姓名”) age: int = Field(…, description=“用户年龄”) email: str = Field(…, description=“用户邮箱”) class UserList(BaseModel): “”“一个包含多个用户的列表”“” users: List[User] # 3. 发起请求,直接指定response_model! completion = client.chat.completions.create( model=“gpt-3.5-turbo”, messages=[ {“role”: “user”, “content”: “提供三个虚构的用户信息,包括姓名、年龄和邮箱。”} ], response_model=UserList, # 核心:指定返回的模型类型 max_retries=2, # 自动重试,提高成功率 ) # 4. 直接得到Pydantic模型实例! extracted_data = completion print(“提取的数据类型:”, type(extracted_data)) print(“\n结构化数据:”) print(extracted_data.model_dump_json(indent=2, ensure_ascii=False)) # 5. 像操作普通对象一样使用数据 for user in extracted_data.users: print(f“用户:{user.name}, 年龄:{user.age}”)5.2 处理复杂嵌套与可选字段
Instructor配合Pydantic能轻松处理复杂场景。
# instructor_advanced.py import instructor from openai import OpenAI from pydantic import BaseModel, Field from typing import List, Optional from enum import Enum class Department(str, Enum): ENGINEERING = “Engineering” SALES = “Sales” HR = “HR” class Address(BaseModel): street: str city: str postal_code: Optional[str] = None # 可选字段 class Employee(BaseModel): id: int full_name: str = Field(…, description=“员工全名”) department: Department email: str address: Optional[Address] = None # 嵌套对象,可选 skills: List[str] = Field(default_factory=list, description=“技能列表”) class CompanyRoster(BaseModel): company_name: str employees: List[Employee] client = instructor.patch(OpenAI()) # 模拟一个复杂的用户请求 user_query = “”” 请为‘创新科技有限公司’生成一个包含5名员工的名单。 需要包含以下信息:员工ID、全名、部门(只能是Engineering, Sales, HR之一)、邮箱。 其中,为至少2名员工添加住址信息(街道、城市、邮编可选)。 为所有员工添加一些相关的技能标签。 “”” roster = client.chat.completions.create( model=“gpt-4-turbo-preview”, # 复杂结构建议使用更强模型 messages=[{“role”: “user”, “content”: user_query}], response_model=CompanyRoster, max_retries=3, ) print(“公司花名册:”) print(roster.model_dump_json(indent=2, ensure_ascii=False)) # 访问嵌套数据 print(f“\n公司名称:{roster.company_name}”) for emp in roster.employees: addr_info = f“, 住址:{emp.address.street}, {emp.address.city}” if emp.address else “” print(f“- {emp.full_name} ({emp.department}){addr_info}”)框架优势总结:
- 声明式编程:用Python类定义结构,无需手动编写JSON Schema。
- 自动重试与校验:框架会自动处理模型的格式错误,并尝试重新生成。
- 类型丰富:完美支持枚举、嵌套模型、可选字段、列表等复杂类型。
- 无缝集成:返回的就是Pydantic模型,便于后续的验证、序列化和使用。
6. 方案四:输出后处理与校验
无论使用哪种方法,在生产环境中添加一道后处理校验都是最佳实践。这可以作为格式错误的最后防线。
6.1 健壮的解析与修复函数
# post_processing.py import json import re from typing import Any, Optional def robust_json_parse(text: str, expected_type: Optional[type] = None) -> Any: “”” 尝试从可能被污染的文本中解析JSON。 步骤:1. 直接解析 2. 清理常见包装 3. 查找JSON子串 4. 尝试修复常见语法错误 “”” original_text = text.strip() # 尝试1:直接解析 try: result = json.loads(original_text) if expected_type and not isinstance(result, expected_type): raise TypeError(f“解析出的类型是 {type(result)}, 但期望的是 {expected_type}”) return result except json.JSONDecodeError as e1: pass # 继续尝试清理 cleaned = original_text # 尝试2:移除Markdown代码块标记 markdown_json_pattern = r‘^```(?:json)?\s*\n?(.*?)\n?```$’ match = re.search(markdown_json_pattern, cleaned, re.DOTALL | re.IGNORECASE) if match: cleaned = match.group(1).strip() # 尝试3:查找第一个‘{‘或’[‘到最后一个’}‘或’]‘之间的内容 # 这可以处理前面有解释性文字的情况 start_chars = {‘{‘: ‘}’, ‘[‘: ‘]’} for start_char, end_char in start_chars.items(): start_idx = cleaned.find(start_char) if start_idx != -1: # 从找到的开始字符开始,找到最后一个匹配的结束字符 stack = [] end_idx = -1 for i in range(start_idx, len(cleaned)): ch = cleaned[i] if ch == start_char: stack.append(ch) elif ch == end_char: if stack: stack.pop() if not stack: # 栈空,找到匹配的结束 end_idx = i break if end_idx != -1: candidate = cleaned[start_idx:end_idx+1] try: result = json.loads(candidate) if expected_type and not isinstance(result, expected_type): raise TypeError(f“子串解析类型不匹配”) return result except: pass # 尝试4:尝试修复一些常见的简单语法错误(谨慎使用) # 例如:单引号替换为双引号(不完全可靠,因为文本中可能包含合法单引号) # 更推荐使用专门的库如 `json5` 或 `demjson3`,这里仅作演示 try: # 这是一个非常简单的修复,可能引入新错误,仅用于最后尝试 repaired = re.sub(r“(?<!\\)‘“, ‘“’, cleaned) # 替换未转义的单引号 repaired = re.sub(r’(?<!\\)’‘, ‘“’, repaired) result = json.loads(repaired) return result except: pass # 所有尝试都失败 raise json.JSONDecodeError(f“无法从文本中解析出有效的JSON。原始文本开头:{original_text[:100]}…”, original_text, 0) # 测试后处理函数 test_cases = [ # 纯JSON ‘[{“name”: “Test”, “age”: 30}]’, # 带Markdown ‘```json\n[{“name”: “Test”, “age”: 30}]\n```’, # 前面有文字 ‘这是你要的数据: [{“name”: “Test”, “age”: 30}]’, # 前后都有文字 ‘结果如下:\n```\n[{“name”: “Test”, “age”: 30}]\n```\n以上是全部信息。’, # 格式略有瑕疵(实际中模型可能产生) “[{‘name’: ‘Test’, ‘age’: 30}]”, # 单引号,非标准JSON ] for i, test in enumerate(test_cases): print(f“\n测试用例 {i+1}: {test[:50]}…”) try: parsed = robust_json_parse(test, expected_type=list) print(f“ 解析成功: {parsed}”) except Exception as e: print(f“ 解析失败: {e}”)6.2 集成到完整流程中
在实际调用中,你应该将后处理作为安全网。
# integrated_pipeline.py import openai import json from post_processing import robust_json_parse # 导入上面的函数 def get_structured_user_data(prompt: str) -> list: “””一个集成了Prompt、调用和后处理的完整流程”“” client = openai.OpenAI() # 使用强化的Prompt enhanced_prompt = f“”” {prompt} 请将输出严格限制为一个纯粹的JSON数组,数组中的每个元素是一个用户对象,包含“name”(字符串)、“age”(整数)、“email”(字符串)字段。 不要输出任何其他文字、标记或解释。 “”” try: response = client.chat.completions.create( model=“gpt-3.5-turbo”, messages=[{“role”: “user”, “content”: enhanced_prompt}], temperature=0.1, # 也可以结合response_format # response_format={“type”: “json_object”}, # 注意:这要求输出是对象,不是数组 ) raw_output = response.choices[0].message.content # 关键步骤:使用健壮的后处理 user_list = robust_json_parse(raw_output, expected_type=list) # 额外的数据校验(可选) for user in user_list: if not isinstance(user.get(‘name’), str): raise ValueError(f“Invalid name type: {user.get(‘name’)}”) if not isinstance(user.get(‘age’), int): # 尝试转换,如果可能 try: user[‘age’] = int(user[‘age’]) except: raise ValueError(f“Invalid age value: {user.get(‘age’)}”) return user_list except (json.JSONDecodeError, ValueError, TypeError) as e: # 记录日志,并可能触发降级策略(如返回空列表、使用默认值、请求人工干预) print(f“结构化数据提取失败: {e}”) # 降级:尝试一个更简单、更直接的请求 return [] # 或 raise # 使用 users = get_structured_user_data(“生成两个用户信息,张三和李四。”) print(“最终获取到的用户列表:”, json.dumps(users, indent=2, ensure_ascii=False))7. 方案对比与选型指南
面对多种方案,如何选择?下表从多个维度进行了对比:
| 特性/方案 | 精炼Prompt工程 | 函数调用 (Function Calling) | Instructor (Pydantic) | 后处理校验 |
|---|---|---|---|---|
| 稳定性 | 中低,依赖模型理解和遵循指令的能力 | 高,模型严格遵循预定义Schema | 非常高,框架提供自动重试和校验 | 作为补充,极高(最后防线) |
| 开发复杂度 | 低,只需编写Prompt | 中,需要定义JSON Schema | 中低,用Python类定义,更直观 | 中,需要编写解析逻辑 |
| 灵活性 | 高,可随时修改Prompt | 中,修改Schema需更新函数定义 | 中,修改Pydantic模型 | 高,可适配各种不规则输出 |
| 类型安全 | 无 | 有,通过Schema定义 | 强,集成Pydantic类型系统 | 无,需自行校验 |
| 输出结构 | 任意JSON | 必须符合函数参数Schema | 必须符合Pydantic模型 | 任意,但目标是规整为预期结构 |
| 适用场景 | 简单、临时的数据提取;对稳定性要求不高的场景 | AI Agent工具调用;需要强格式保证的API交互 | 复杂嵌套数据提取;需要类型安全和自动重试的项目 | 必备的安全生产环节;处理第三方或不可控模型的输出 |
| 推荐指数 | ⭐⭐ | ⭐⭐⭐⭐⭐ (Agent场景) | ⭐⭐⭐⭐⭐ (数据提取场景) | ⭐⭐⭐⭐⭐ (必须搭配使用) |
选型建议:
- 对于AI Agent开发:优先使用函数调用(Function Calling)。这是OpenAI为工具使用设计的原生方式,稳定性和生态兼容性最好。
- 对于复杂数据提取任务:强烈推荐使用Instructor库。它用Pythonic的方式将定义、调用、校验融为一体,大幅提升开发效率和可靠性。
- Prompt工程:作为辅助手段,在函数调用或Instructor的提示部分使用,进一步明确任务要求。
- 后处理校验:无论采用哪种方案,都必须实施。这是保证程序鲁棒性的关键。
8. 高级技巧与面试常见问题
8.1 处理模型“幻觉”与字段缺失
即使格式正确,模型生成的内容也可能不符合要求(幻觉)或缺失字段。
解决方案:
- 在Schema/模型定义中使用
Field(…, description=“”):提供清晰、无歧义的字段描述。 - 设置
required字段:在JSON Schema或Pydantic中明确必填字段。 - 使用
max_retries(Instructor支持):让框架自动重试。 - 后验证与默认值:
from pydantic import BaseModel, Field, validator class User(BaseModel): name: str age: int = Field(default=0, ge=0, le=120) # 设置默认值和范围 email: Optional[str] = None @validator(‘email’) def validate_email_format(cls, v): if v is not None and “@” not in v: # 可以尝试修复,或引发错误 # 或者 return a default like “unknown@example.com” raise ValueError(‘invalid email format’) return v
8.2 流式输出中的JSON处理
当使用流式响应(Streaming)时,不能等完整响应回来再解析。
策略:
- 对于函数调用:OpenAI的流式响应中,
tool_calls的arguments是一个增量生成的字符串。你需要自己拼接这些Delta,并在收到结束信号后尝试解析。 - 使用专门模式:有些服务或框架提供了“流式JSON”模式,如
response_format={“type”: “json_object”}在某些模型上支持流式输出JSON的每个键值对。
8.3 大模型面试高频考点
如何保证大模型输出JSON的稳定性?
- 标准答案:采用多层保障策略。首选使用模型的函数调用(Function Calling)功能,通过预定义严格的JSON Schema来约束输出;其次,可以使用像Instructor这样的库,通过Pydantic模型声明数据结构,并利用其自动重试机制。同时,必须编写健壮的后处理解析函数,以处理模型可能输出的非纯JSON文本(如Markdown包装)。在Prompt设计上,要使用清晰、强制的指令,并结合少样本示例。
函数调用(Function Calling)的原理是什么?
- 模型并不直接执行函数。开发者预先定义好工具(函数)的列表及其参数Schema。当用户输入与某个工具的描述匹配时,模型会输出一个特殊的结构化消息,表明它“想要调用”某个函数,并生成一个符合该函数Schema的JSON参数。开发者收到这个请求后,在自己的代码中真正执行对应的函数。
如果模型就是不生成有效JSON怎么办?
- 降级策略:首先记录错误并重试(2-3次)。如果仍然失败,可以回退到一个更简单、约束更强的Prompt。最终手段是向用户返回一个友好的错误信息,并提示其重新表述请求或转为人工处理。
- 监控与迭代:收集所有失败的案例,分析原因。是Prompt不清晰?Schema太复杂?还是模型能力不足?根据分析结果优化你的Schema和Prompt。
如何设计一个用于提取信息的Pydantic模型?
- 从核心实体开始,使用明确的字段名和类型(
str,int,List,Optional)。 - 为每个字段添加
Field(…, description=“”),描述要具体、无歧义。 - 使用嵌套模型(
BaseModel)组织复杂关系。 - 利用枚举(
Enum)约束字段的取值范围。 - 为可选字段设置合理的默认值(如
None)。 - 添加验证器(
@validator)进行业务逻辑校验。
- 从核心实体开始,使用明确的字段名和类型(
9. 完整实战案例:构建一个稳定的用户信息提取Agent
让我们综合运用以上知识,构建一个从一段自由文本中提取用户信息并存入模拟数据库的简单Agent。
# user_info_agent.py import instructor from openai import OpenAI from pydantic import BaseModel, Field, validator from typing import List, Optional import json import sqlite3 from datetime import datetime # --- 1. 定义数据结构 --- class Address(BaseModel): street: Optional[str] = None city: Optional[str] = None country: str = “中国” # 默认值 class UserInfo(BaseModel): “”“从文本中提取出的单个用户信息”“” name: str = Field(…, description=“用户的全名”) age: Optional[int] = Field(None, ge=0, le=120, description=“用户年龄,如果没有明确提及则为None”) email: Optional[str] = Field(None, description=“邮箱地址”) phone: Optional[str] = Field(None, description=“手机号码”) address: Optional[Address] = None source_text_snippet: str = Field(…, description=“原文中提及该用户的那部分文本”) @validator(‘email’) def email_contains_at(cls, v): if v is not None and “@” not in v: # 简单校验,生产环境应用更复杂的正则 raise ValueError(‘Email must contain @’) return v class ExtractedData(BaseModel): “”“提取任务的总结果”“” users: List[UserInfo] raw_text_summary: str = Field(…, description=“对原始文本的简要总结”) # --- 2. 修补客户端并定义提取函数 --- client = instructor.patch(OpenAI()) def extract_users_from_text(text: str) -> ExtractedData: “””核心提取函数,使用Instructor”“” prompt = f“”” 请从以下文本中提取所有提到的用户信息。 文本内容: “”{text}”” 请仔细识别文中提到的每一个人,并尽可能提取他们的姓名、年龄、联系方式(邮箱或电话)和住址信息。 如果某些信息(如年龄、邮箱)没有明确提及,请将其设为null。 请确保“source_text_snippet”字段准确记录原文中描述该用户的句子或片段。 最后,请用一句话总结原始文本的主要内容,填入“raw_text_summary”。 “”” try: extracted = client.chat.completions.create( model=“gpt-4-turbo-preview”, # 复杂信息提取建议用更强模型 messages=[{“role”: “user”, “content”: prompt}], response_model=ExtractedData, max_retries=3, temperature=0, ) return extracted except Exception as e: print(f“信息提取失败: {e}”) # 返回一个空的提取结果作为降级 return ExtractedData(users=[], raw_text_summary=“提取失败”) # --- 3. 模拟数据库存储 --- def init_database(): conn = sqlite3.connect(‘:memory:’) # 内存数据库,方便演示 cursor = conn.cursor() cursor.execute(“”” CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER, email TEXT, phone TEXT, address_street TEXT, address_city TEXT, address_country TEXT, source_snippet TEXT, extracted_at TIMESTAMP, raw_summary TEXT ) “””) conn.commit() return conn def save_to_database(conn, data: ExtractedData): cursor = conn.cursor() for user in data.users: cursor.execute(“”” INSERT INTO users (name, age, email, phone, address_street, address_city, address_country, source_snippet, extracted_at, raw_summary) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) “””, ( user.name, user.age, user.email, user.phone, user.address.street if user.address else None, user.address.city if user.address else None, user.address.country if user.address else “中国”, user.source_text_snippet, datetime.now().isoformat(), data.raw_text_summary )) conn.commit() print(f“成功保存 {len(data.users)} 条用户记录到数据库。”) # --- 4. 主程序 --- if __name__ == “__main__”: # 模拟输入文本 input_text = “”” 在我们的项目团队中,张三(25岁)主要负责后端开发,他的联系邮箱是zhangsan@company.com。 李四来自北京,住在朝阳区建国路123号,她今年30岁了,电话是13800138000。 还有一位同事王五,他的邮箱是wangwu@example.org,常驻上海。 最近我们招募了实习生赵六,他今年22岁。 “”” print(“原始文本:”) print(input_text) print(“\n” + “=”*50 + “\n”) # 初始化数据库 db_conn = init_database() # 执行提取 print(“正在使用大模型提取用户信息…”) result = extract_users_from_text(input_text) print(“\n提取结果:”) print(result.model_dump_json(indent=2, ensure_ascii=False)) # 保存到数据库 if result.users: save_to_database(db_conn, result) # 查询验证 cursor = db_conn.cursor() cursor.execute(“SELECT name, age, email, phone FROM users”) saved_users = cursor.fetchall() print(“\n数据库中保存的用户:”) for user in saved_users: print(user) else: print(“未提取到用户信息。”) db_conn.close()这个案例展示了从定义数据结构、使用Instructor稳定提取、数据后验证到持久化存储的完整流程,是一个生产可用Agent的简化原型。
10. 总结与最佳实践清单
要确保大模型稳定输出JSON,没有银弹,而是一套组合拳。以下是贯穿整个开发周期的核心实践:
设计阶段
- 明确需求:首先彻底厘清你需要的数据结构。使用工具如
pydantic或JSON Schema进行正式定义。 - 选择正确工具:对于Agent工具调用,用原生函数调用;对于复杂数据提取,用
Instructor+Pydantic。
- 明确需求:首先彻底厘清你需要的数据结构。使用工具如
开发阶段
- 强化Prompt:在系统或用户消息中明确要求“只输出JSON”,并提供清晰示例。将
temperature设为较低值(如0-0.3)。 - 利用API特性:如果适用,务必使用
response_format={ “type”: “json_object” }参数。 - 实现健壮解析:编写一个
robust_json_parse函数,作为处理模型原始输出的安全网。
- 强化Prompt:在系统或用户消息中明确要求“只输出JSON”,并提供清晰示例。将
测试与监控阶段
- 全面测试:使用包含边缘案例(缺失字段、格式污染、怪异字符)的文本测试你的流程。
- 实施重试机制:对于非关键任务,配置2-3次自动重试(如Instructor的
max_retries)。 - 记录失败案例:所有解析失败的请求,都应记录其原始Prompt和模型输出,用于后续分析和Prompt/Schema优化。
生产部署阶段
- 设置超时与降级:为LLM调用设置合理超时,并规划好降级策略(如返回空值、使用缓存、触发人工流程)。
- 监控关键指标:跟踪JSON解析成功率、字段填充率、模型调用延迟和成本。
- 持续迭代:大模型能力和最佳实践在快速演进,定期回顾和更新你的Prompt、Schema及后处理逻辑。
稳定、可靠的结构化输出是将大模型从“玩具”变为“生产工具”的关键一步。通过本文介绍的多层策略,你可以显著提升AI应用的数据交互可靠性,为构建复杂的智能体系统和自动化流程打下坚实基础。