1. 项目概述:为什么我们需要Structured Output?
在Agent开发领域,让AI模型输出结构化数据(如JSON)已经成为现代智能体系统的标配能力。传统的大语言模型输出就像个话痨朋友——回答冗长、格式随意,需要开发者手动解析关键信息。而Structured Output技术相当于给模型戴上了"数据模具",强制其按照预定格式输出规整的数据结构。
最近我在开发一个电商客服Agent时深有体会:当用户询问"对比iPhone 15和三星S23的摄像头参数"时,理想输出应该是:
json复制{
"products": [
{
"name": "iPhone 15",
"camera_specs": {
"main_camera": "48MP",
"ultra_wide": "12MP",
"features": ["Night mode", "Deep Fusion"]
}
},
{
"name": "Samsung S23",
"camera_specs": {
"main_camera": "50MP",
"ultra_wide": "12MP",
"features": ["Space Zoom", "Director's View"]
}
}
]
}
而不是一段需要正则表达式提取的散文式回答。这就是Structured Output的核心价值——让机器对话更机器可读。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 核心技术解析:JSON Schema驱动输出
2.1 JSON Schema基础规范
JSON Schema相当于给JSON数据制定的"宪法",通过定义字段类型、格式约束、必填项等规则,确保模型输出符合预期结构。以下是一个典型的产品对比Schema:
json复制{
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {
"products": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"camera_specs": {
"type": "object",
"properties": {
"main_camera": {"type": "string"},
"ultra_wide": {"type": "string"},
"features": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["main_camera"]
}
},
"required": ["name"]
}
}
},
"required": ["products"]
}
关键约束点:
type定义字段数据类型required声明必填字段items规定数组元素结构- 支持嵌套对象定义
2.2 主流框架实现方案
2.2.1 LangChain实现
通过StructuredOutputParser组件实现:
python复制from langchain.output_parsers import StructuredOutputParser
from langchain.prompts import PromptTemplate
schema = {
"products": [{
"name": "string",
"price": "float",
"in_stock": "boolean"
}]
}
parser = StructuredOutputParser.from_response_schemas(schema)
format_instructions = parser.get_format_instructions()
prompt = PromptTemplate(
template="列出3个{product_type}及其价格\n{format_instructions}",
input_variables=["product_type"],
partial_variables={"format_instructions": format_instructions}
)
2.2.2 LlamaIndex方案
使用Pydantic模型定义结构:
python复制from pydantic import BaseModel
from typing import List
class Product(BaseModel):
name: str
price: float
features: List[str]
response = index.query(
"比较iPhone和三星旗舰机",
output_cls=Product
)
3. 实战开发全流程
3.1 需求定义阶段
以电商价格监控Agent为例:
-
明确数据结构需求:
- 商品名称(字符串)
- 当前价格(浮点数)
- 历史最低价(浮点数)
- 促销标签(字符串数组)
- 库存状态(布尔值)
-
设计Schema验证规则:
- 价格必须为正数
- 促销标签最多5个
- 必须包含商品名称
3.2 提示词工程技巧
在系统提示中加入结构化引导:
code复制你是一个专业的价格监控AI,请严格按照以下JSON格式输出:
{
"products": [{
"name": "商品名称",
"current_price": 当前价格,
"lowest_price": 历史最低价,
"promotion_tags": ["促销标签1", "标签2"],
"in_stock": 库存状态
}]
}
注意:
- 价格保留两位小数
- 促销标签不超过5个
- 必须验证库存状态
3.3 代码实现示例
3.3.1 Python完整示例
python复制import json
from typing import List
from pydantic import BaseModel, validator
class Product(BaseModel):
name: str
current_price: float
lowest_price: float
promotion_tags: List[str] = []
in_stock: bool
@validator('promotion_tags')
def validate_tags(cls, v):
if len(v) > 5:
raise ValueError("最多5个促销标签")
return v
def parse_response(response_text: str) -> List[Product]:
try:
data = json.loads(response_text)
return [Product(**item) for item in data["products"]]
except Exception as e:
print(f"解析失败: {str(e)}")
raise
# 模拟LLM输出
llm_output = """
{
"products": [
{
"name": "iPhone 15",
"current_price": 6999.00,
"lowest_price": 6499.00,
"promotion_tags": ["暑期特惠", "12期免息"],
"in_stock": true
}
]
}
"""
products = parse_response(llm_output)
print(products[0].name) # 输出: iPhone 15
3.3.2 Java实现方案
java复制import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.List;
public class Product {
private String name;
private double currentPrice;
private double lowestPrice;
private List<String> promotionTags;
private boolean inStock;
// Getters and Setters
@JsonProperty("name")
public String getName() { return name; }
@JsonProperty("current_price")
public double getCurrentPrice() { return currentPrice; }
// 验证逻辑
public void validate() throws IllegalArgumentException {
if (promotionTags != null && promotionTags.size() > 5) {
throw new IllegalArgumentException("促销标签不能超过5个");
}
}
}
// 使用Jackson解析
ObjectMapper mapper = new ObjectMapper();
Product[] products = mapper.readValue(jsonString, Product[].class);
for (Product p : products) {
p.validate();
}
4. 避坑指南与性能优化
4.1 常见问题排查表
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| JSON解析失败 | 模型输出包含非JSON文本 | 使用json.loads(text.split("```json")[1].split("```")[0])提取 |
| 字段缺失 | Schema未标记required | 在Schema中明确required字段 |
| 类型错误 | 模型输出字符串而非数字 | 提示词明确要求"price应为数字类型" |
| 数组越界 | 未限制数组长度 | 在Schema中添加maxItems约束 |
4.2 性能优化技巧
-
Schema精简原则:
- 嵌套不超过3层
- 单个Schema字段不超过15个
- 优先使用基本类型(string/number/boolean)
-
缓存解析器实例:
python复制# 错误做法:每次请求新建解析器 def process_query(query): parser = StructuredOutputParser(...) # ... # 正确做法:全局缓存 cached_parser = StructuredOutputParser(...) -
流式处理大响应:
python复制import ijson def parse_large_response(stream): for item in ijson.items(stream, "products.item"): yield Product(**item)
5. 高级应用场景
5.1 动态Schema适配
根据用户查询自动调整输出结构:
python复制def generate_dynamic_schema(query):
if "对比" in query:
return COMPARISON_SCHEMA
elif "推荐" in query:
return RECOMMENDATION_SCHEMA
else:
return DEFAULT_SCHEMA
5.2 多Agent数据交换
当Agent需要传递结构化数据时:
json复制{
"metadata": {
"sender": "price_monitor",
"timestamp": "2024-03-20T14:30:00Z"
},
"payload": {
"alert_type": "price_drop",
"products": [...]
}
}
5.3 结合OpenAPI规范
将JSON Schema转化为API文档:
yaml复制paths:
/product-info:
post:
responses:
'200':
content:
application/json:
schema:
type: array
items:
$ref: '#/components/schemas/Product'
components:
schemas:
Product:
type: object
properties:
name:
type: string
price:
type: number
format: float
6. 测试验证策略
6.1 单元测试示例
python复制import pytest
def test_product_parsing():
test_data = {
"products": [{
"name": "Test",
"current_price": 100.0,
"lowest_price": 90.0,
"in_stock": True
}]
}
products = parse_response(json.dumps(test_data))
assert products[0].name == "Test"
assert products[0].current_price == 100.0
def test_invalid_price():
with pytest.raises(ValueError):
test_data = {"products": [{"name": "Test", "current_price": -100}]}
parse_response(json.dumps(test_data))
6.2 模糊测试方案
使用hypothesis生成随机测试数据:
python复制from hypothesis import given, strategies as st
@given(
st.lists(
st.fixed_dictionaries({
"name": st.text(min_size=1),
"current_price": st.floats(min_value=0),
"lowest_price": st.floats(min_value=0)
}),
min_size=1
)
)
def test_random_products(products):
data = {"products": products}
result = parse_response(json.dumps(data))
assert len(result) == len(products)
在实际项目中,我发现这些技术组合使用可以降低约70%的数据清洗工作量。特别是在构建Agent工作流时,结构化的输出就像给每个Agent安装了标准化的数据接口,让机器协作变得异常顺畅。
