本次实践以一个真实的推荐系统数据工程场景为背景：把 210 段风格自由、毫无 schema 可言的餐厅描述文本，批量转换成机器可查询的结构化 JSON 知识库。完整走一遍One-shot 提示抽取 → Pydantic 模式验证 → 错误回喂自动修复的三段式管道，最终 210 条输入产出 209 条合法 JSON——看看 LLM 结构化输出的工程化闭环到底怎么做，以及"让 LLM 自己修自己"这个看似偷懒的设计，为什么在工程上反而说得通。
一、问题：非结构化文本是搜索的天敌
场景不复杂：一家 AI 公司要做美食推荐产品Connoisseur Companion，不满足于传统的 1–5 星评分，想从评论文本里挖出餐厅的氛围、饮食限制和"招牌菜"。手里的原料是一份 73KB 的 TXT 文件——210 段餐厅描述，每段长这样：
The Gilded Artichokebrings abohemian chicenergy to the hills ofSilver Lake, operating as anupscale bistrothat prioritizesFarm-to-Table Californianingredients… Price range: $$$$
人读起来信息密度很高，但对程序来说这是灾难：
- 没有统一 schema：菜系、评分、价格散落在句子的任何位置，正则写不完；
- 类型不统一：评分是
4.5/5，价格是$$$$，招牌菜是一句话里的短语； - 无法索引：想查"洛杉矶 300 元以内、有素食选项的融合菜"，关键词匹配基本靠缘分。
这正是 LLM 结构化抽取的典型战场：用自然语言理解替代正则规则，用模式验证保证输出可控。整个管道的架构如下：
注意最后那条回边——验证失败不是终点，而是修复循环的起点。这是全文最重要的一处设计，后面细讲。
二、数据切分
文本按空行分段，加载和切分只需要三行：
file_path="California-Culinary-Map.txt"withopen(file_path,'r')asfile:data=file.read()restaurant_list=data.split('\n\n')# 按空行切段restaurant_list=restaurant_list[1:]# 第一段是标题，去掉len(restaurant_list)得到 210。这里有个不起眼但值得强调的点：切段就是建立处理单元的边界契约。后续所有环节（抽取、验证、修复、入库）都以"段"为原子单位，任何一段失败都不应污染其他段。边界划得干净，后面的容错设计才有立足点——如果一段里混进两家餐厅，LLM 抽取和验证都会陷入混沌。
三、封装 LLM：面向 OpenAI 兼容协议编程
模型调用层用 LangChain 的ChatOpenAI封装，好处是换模型只改三个参数：
fromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportSystemMessage,HumanMessagedefllm_model(system_msg,prompt_txt):model=ChatOpenAI(model="mimo-v2.6-flash",base_url="https://token-plan-cn.xiaomimimo.com/v1",api_key="...",)messages=[SystemMessage(content=system_msg),HumanMessage(content=prompt_txt)]response=model.invoke(input=messages)returnresponse.contentbase_url+model+api_key三件套一换，实验里从 CSDN 的deepseek-v4-flash自由切换到小米的mimo-v2.6-flash。
在 OpenAI 兼容协议已经成为事实标准的今天，业务代码不应该绑定任何一家模型供应商——这对生产环境尤其重要。
另外套了一层带重试的包装，网络抖动时不至于让 210 次循环中途夭折：
defsafe_llm_call(system_msg,prompt_txt,retries=3):foriinrange(retries):try:returnllm_model(system_msg,prompt_txt)exceptException:time.sleep(2)return"Failed after retries"四、One-shot 提示：把 schema 塞进示例里
1. 先明确要抽什么
从每段描述中提取 9 个字段：
| 字段 | 含义 | 原文形态 | 目标类型 |
|---|---|---|---|
name | 餐厅名 | **The Gilded Artichoke** | str |
location | 位置 | Silver Lake | str |
type | 餐厅类型 | upscale bistro | str |
food_style | 餐饮风格 | Farm-to-Table Californian | str |
rating | 评分 | 4.5/5 | float |
price_range | 价格区间 | $$$$ | int（美元符号个数） |
signatures | 招牌菜 | 句中短语 | List[str] |
vibe | 氛围 | bohemian chic | str |
environment | 环境描述 | 一句话 | str |
shortcomings | 不足 | 可能缺失 | List[str] |
其中price_range: $$$$ → 4是典型的隐式类型转换需求——人一眼看懂，但如果不明确指示，LLM 十有八九原样输出"$$$$"字符串。
2. One-shot 模板的设计
不用空口描述格式，直接给一对完整的输入-输出示例：
EXAMPLE_OUTPUT=""" { "name": "Mar de Cortez", "location": "Santa Monica", "type": "casual taqueria", "food_style": "Baja-style seafood", "rating": 4.2, "price_range": 1, "signatures": ["beer-battered snapper tacos", "zesty octopus ceviche"], "vibe": "salt-air energy", "environment": "a premier sun-drenched spot for open-air dining near the pier.", "shortcomings": [] } """defrestaurant_data_structure_prompt_generation(restaurant_paragraph):base_system_msg="你是一个专业的餐厅数据结构化处理助手。"base_user_prompt=f""" Task: 将餐厅描述结构化处理, 输出JSON格式。 Restaurant description:{restaurant_paragraph}Example: Input Restaurant Description:{EXAMPLE_RESTAURANT_PARAGRAPH}Output:{EXAMPLE_OUTPUT}"""returnbase_system_msg,base_user_prompt实测单条抽取的输出：
{"name":"The Gilded Artichoke","location":"Silver Lake","type":"upscale bistro","food_style":"Farm-to-Table Californian","rating":4.5,"price_range":4,"signatures":["lavender-rubbed roasted chicken","delicate heirloom tomato tarts"],"vibe":"bohemian chic","environment":"a high-end greenhouse feel with reclaimed wood and floor-to-ceiling windows.","shortcomings":[]}字段一个不少，$$$$正确转成了4，招牌菜准确抓出了两个。One-shot 的本质是把"格式规范"从抽象的文字描述变成具体的对齐目标——模型对示例的模仿能力远强于对指令的遵循能力，这也是 few-shot 提示在结构化输出场景几乎必用的原因。
五、Pydantic 验证：schema 即契约
LLM 输出"看起来像 JSON"不等于"是合法且符合业务的 JSON"。用 Pydantic 把数据契约正式化：
frompydanticimportBaseModel,Field,ValidationErrorfromtypingimportList,OptionalclassRestaurant(BaseModel):name:strlocation:strtype:strfood_style:strrating:Optional[float]=Noneprice_range:Optional[int]=Nonesignatures:List[str]=Field(default_factory=list)vibe:Optional[str]=Noneenvironment:strshortcomings:List[str]=Field(default_factory=list)restaurant_data=Restaurant.model_validate_json(json_data=test_response)这个 schema 有两层设计考量：
- 核心字段非空（
name/location/type/environment是 str 必填）：缺了就没法做检索，宁缺毋滥； - 易缺失字段 Optional + 默认值（
rating、shortcomings等）：原文里可能压根没提，硬性要求反而制造大量验证失败。
model_validate_json一句话同时完成JSON 语法校验 + 字段类型校验 + 缺失字段校验。更重要的是，它抛出的ValidationError自带结构化的错误详情（哪个字段、什么原因）——这份错误信息，正是下一环节自动修复的燃料。
六、自动修复：让 LLM 当自己的"格式审查员"
1. 为什么需要修复环节
批处理 210 条时，单条 95% 的成功率意味着期望失败 10 条左右。失败原因五花八门：字段名拼错、价格输出了字符串、signatures里混进了整句话。靠人工盯着改，210 条能把人改疯。
实验给出的方案是引入第二个 LLM 角色——修复专家，输入是"错误输出 + 验证错误详情"，输出是修好的 JSON：
defJSON_auto_repair_prompts(candidate_json_output,error_message):auto_repair_system_msg=f""" 这是原始的错误输出：{candidate_json_output}; 这是错误信息：{error_message}"""auto_repair_prompt=f""" 根据错误信息{error_message}和原始非结构化脚本，修复{candidate_json_output}文件。 """returnauto_repair_system_msg,auto_repair_prompt注意error_message用的是 Pydantic 的str(e.errors())——把结构化的错误详情（字段路径 + 违反原因）喂给修复模型，而不是只说"格式错了"，修复的命中率完全不在一个量级。
2. 生成-验证-修复闭环
主循环把三个环节串起来：
structured_restaurant_lists=[]fori,restaurant_paragraphinenumerate(restaurant_list):# 1. 生成候选 JSONbase_system_msg,base_user_prompt=restaurant_data_structure_prompt_generation(restaurant_paragraph=restaurant_paragraph)raw_response=llm_model(system_msg=base_system_msg,prompt_txt=base_user_prompt)candidate_json_output=raw_response.replace("```json","").replace("```","").strip()# 2. 验证 + 自动修复（限次）max_attempts=1# 防止无限循环current_attempt=0restaurant_data=Nonewhilecurrent_attempt<max_attempts:try:restaurant_data=Restaurant.model_validate_json(candidate_json_output)breakexceptValidationErrorase:current_attempt+=1auto_repair_system_msg,auto_repair_prompt=JSON_auto_repair_prompts(candidate_json_output,error_message=f"{str(e.errors())}")candidate_json_output=llm_model(system_msg=auto_repair_system_msg,prompt_txt=auto_repair_prompt)# 3. 成功才入库，失败告警跳过ifrestaurant_dataisnotNone:structured_restaurant_lists.append(restaurant_data)else:print(f"⚠️ 第{i+1}/{len(restaurant_list)}条餐厅数据修复失败，已跳过")if(i+1)%20==0:print(f'{i+1}out of{len(restaurant_list)}is done')print('ALL DONE!!')跑完的实际结果：210 条输入，209 条入库，第 1 条修复失败被跳过。
3. 这个设计里藏着三个工程决策
① 为什么要限次（max_attempts=1）？修复 LLM 也是 LLM，不是确定性程序——修复失败后再喂回去重试，可能陷入"生成错 → 修复 → 还是错"的死循环，而每次循环都是真金白银的 token。限次 + 跳过，用完成率换确定性。
② 为什么跳过而不是中断？批处理的第一原则是单点失败不拖垮全局。209 条质量合格的数据，价值远大于 210 条里夹着一条手工兜底的脏数据。跳过的记录留下了告警日志，可以事后人工补。
③ "让 LLM 修 LLM"是不是最佳实践？说实话不是。生产级方案有更严格的约束手段：JSON mode / function calling 强制输出格式、解码层约束（如 outlines、guidance）、独立的校验重试框架。但在本场景中，抽取和修复对模型来说都是简单任务，成功率极高，引入重型框架的收益抵不上复杂度成本。工程选型的标准从来不是"理论最优"，而是"够用且最简"。
七、入库：两个容易被忽略的收尾细节
structured_restaurant_lists_json=[r.__dict__forrinstructured_restaurant_lists]fori,responseinenumerate(structured_restaurant_lists_json):response['itemId']=1000001+i filename='structured_restaurant_data.json'withopen(filename,'w',encoding='utf-8')asf:json.dump(structured_restaurant_lists_json,f,indent=4)__dict__转换：Pydantic 对象不能直接json.dump，先转成字典；itemId对齐：分配1000001 + i的连续 ID，与后续用户评论数据集里的itemId对齐——这一步是在为多表关联预埋主键。结构化的最终目的不是存文件，而是让餐厅知识库和用户行为数据能够 JOIN 起来做推荐。schema 设计要往前看一步。
八、结语：这套管道的通用性
回头看，这个实验给出的不是"餐厅数据处理方案"，而是一个可迁移的 LLM 结构化抽取模式：
变的是数据和 schema，不变的是四条原则：
- 示例优于指令——想稳定拿到结构，就给它看一个结构；
- 验证先于信任——LLM 的输出在通过 schema 校验前，一律视为候选；
- 错误是资源——结构化的验证错误详情，是修复环节最精准的提示词；
- 提示词非常关键——好的提示词会大大提高执行效率，同时减少Token消耗；
- 批处理要有退路——限次、跳过、告警，用局部损失换全局确定。
这套管道的下一步，是把图片和 URL 也纳入知识库——用多模态模型给菜谱图片生成描述，让"视觉信息"变成可检索的文本。
