企业官网建设流程全解析

做RAG项目的时候，我发现自己大部分时间不是耗在embedding或者向量库上，而是在跟模型的输出较劲。尤其是当用户要的是一张数据表，而模型却给你甩来一大段自然语言时，那感觉真是“检索做得再好，解析一塌糊涂”。这次要聊的，就是我们内部这个叫Data-Processor的模块下的一个子任务：output_parsing01，核心就一句话——把RAG返回的非结构化结果，解析成可用的结构化数据，最终以DataFrame的形式落到下游。这篇文章会把整个方案的来龙去脉、设计取舍、踩坑记录都摊开来讲，适合正在做RAG管线、或者被模型输出格式折腾得不轻的开发者参考。

1. 为什么要在RAG后面单独做一个输出解析模块

1.1 先看清楚业务卡在哪

RAG类应用走完“检索-上下文拼接-生成”之后，模型返回的是自然语言。这在纯聊天场景没问题，但在数据分析、报表生成、内部知识库管理等场景里，用户真正想要的是能直接进Excel、能进数据库表格的东西。举个例子，我们做个文档问答，用户问“过去三个月有哪些订单金额超过5000元”，模型也许能正确回答出三五笔订单，但把订单号、金额、日期混在一起写成一段话，下游根本没法自动处理。

更麻烦的是，同一个问题在不同文档库环境下，答案格式也可能各不相同。今天模型给你列了个编号列表，明天可能变成一段带冒号的描述，后天又给你来了个Markdown表格。如果每次都靠正则硬匹配，那就彻底沦为“针对一个case写一套规则”的体力活。于是我们意识到，RAG管道里缺一个专门的输出解析层，负责把模型输出“洗”成规整结构。

1.2 为什么我选JSON中转而不是直接生成表格

这里有个重要的设计决定：要不要直接让模型输出一个Markdown表格，或者直接输出CSV文本？

我自己的实际感受是，多轮测试下来，让模型直接输出CSV或者表格式文本，解析成本不高，但纠错成本很高。因为表格文本的格式非常脆弱，少一个逗号、多一个换行都会导致解析偏掉。而JSON是个具有明确层级和标准库支持的格式，几乎每种语言都有现成的解析器，模型对JSON的生成能力也训练得特别充分。

所以output_parsing01这个子模块的设计思路，不是做“文本转表格”，而是先要求模型把答案整理成JSON，再由解析器将JSON平铺转换成DataFrame。表面上看多了一步，实际上是把不稳定因素收敛到“JSON合法性”这一个点上，后续的字段映射、类型转换都有清晰的落点。

1.3 模块边界怎么切

最初我们也走过后端服务里到处塞解析逻辑的弯路，后来发现解析逻辑一旦散落各处，模型输出格式一调整就要改一堆地方。于是我们把整块内容拆成了一个独立阶段：

• 负责检索、生成的主流程不管输出格式；
• output_parsing01只接收“模型回答字符串”和“字段定义列表”；
• 返回结果统一为DataFrame对象。

这样定义之后，模型输出的prompt怎么调整、解析器内部怎么升级，互不干扰。外部调用方依然只认DataFrame格式，内部使用方也不用关心下游怎么消化。

2. 从原始回答到DataFrame的数据流设计

2.1 pipeline完整链路

我先把当前项目的整条链路画个简化版，方便理解output_parsing01处在什么位置：

1. 用户输入问题；
2. 检索模块从向量库取出相关文档片段；
3. prompt组装模块把用户问题+检索片段+输出格式要求拼好；
4. 模型生成回答；
5. 拿到原始字符串后，进入output_parsing01；
6. 解析器清洗、转换、组装成DataFrame；
7. 写入业务表或直接返回给前端展示。

从这个顺序里能看到，解析模块其实是RAG服务的最后一公里。它决定了前面检索、生成这些工作最终能不能转化成一个可消费的产品功能。这一公里做不好，再强的检索能力展示出来也是一段不可用的文本。

2.2 让模型输出稳定结构的三板斧

既然要让JSON中转，那怎么保证模型每次都能输出合法、且符合我们字段定义的JSON？我试下来有三个办法最有效。

第一，写死字段名。不能只告诉模型“返回订单信息”，要明确列出字段名、类型、示例值。比如“订单号：字符串，日期：YYYY-MM-DD格式，金额：数字，单位元”。字段名都提前固定，后面解析器才能精确匹配。

第二，强制标记JSON区块。在prompt中让模型只输出JSON代码块，不要输出多余解释语言。有些模型偶尔会在JSON前后加“好的，我已经帮你整理好了”，这些杂讯很容易干扰直接解析。所以我们在prompt里明确写“直接输出json，不要任何说明或Markdown标记之外的文字”。

第三，开启保守生成参数。生成阶段把temperature调低，比如设为0.1甚至0。尤其在输出解析场景里，低温度能明显降低字段名被改写、数字被胡乱格式化的概率。

我自己实测下来，这三板斧配合使用，基本能把合法JSON的产出比例从七八成拉高到九成以上。但依旧不能做到百分之百，所以解析器一定要具备容错能力，这个我们后面重点讲。

2.3 解析器输入输出与下游对接

output_parsing01的输入，其实就是两个东西：模型返回的原始字符串、用户或开发者定义的字段schema。

输出是DataFrame，但这里有个关键细节：解析器不能假设所有模型回答都天然包含全部字段。比如模型漏答了某个字段，解析器要能补上默认值；模型多返回了某个字段，解析器要有策略决定是丢弃还是纳入附加列。

在与下游对接上，我们定义了一个统一规范：解析后的DataFrame列名与schema中的字段名一一对应，且类型已经转换到位。这样下游不管是写数据库还是渲染到前端，都不需要再关心原始文本是什么样的。

3. 核心代码实现：JSON转DataFrame的正确姿势

3.1 定义字段模型：字段名、类型、默认值

我先说下我们的做法，用一个列表来描述目标结构，每项包含字段名、数据类型和缺失时的默认值。不用完整引入数据验证框架，是因为这个模块本身就是轻量级的，不想为了一个小环节背上一堆依赖。

# field_schema.py FIELDS = [ {"name": "order_id", "type": "str", "default": ""}, {"name": "order_date", "type": "date", "default": None}, {"name": "amount", "type": "float", "default": 0.0}, ]

这样定义的好处是，解析器在拿到模型输出时，能清楚地知道每个字段的可接受形态。比如order_date如果模型返回的是字符串“2024-06-01”，解析器就知道要转成datetime对象；如果返回空，那就用默认值None。

3.2 解析函数：容错、兜底、类型转换

这里我直接给出一个核心函数，涵盖了我们目前80%的使用场景。第一步提取JSON区块，第二步做json.loads解析，第三步按字段定义遍历转换，第四步汇总成DataFrame。

import json import re import pandas as pd from datetime import datetime def extract_json_block(raw_text: str) -> str: # 优先匹配 ```json ... ``` 代码块 pattern = r"```json\s*(.*?)\s*```" match = re.search(pattern, raw_text, re.DOTALL) if match: return match.group(1) # 退而求其次，匹配第一个 { 到最后一个 } start = raw_text.find("{") end = raw_text.rfind("}") if start != -1 and end != -1 and end > start: return raw_text[start:end + 1] return raw_text def safe_parse_record(record: dict, fields: list) -> dict: result = {} for field in fields: name = field["name"] ftype = field["type"] default = field["default"] value = record.get(name, default) if value is None or value == "": result[name] = default continue try: if ftype == "date": result[name] = datetime.strptime(str(value).strip(), "%Y-%m-%d") elif ftype == "float": result[name] = float(str(value).replace(",", "").replace("元", "").strip()) elif ftype == "int": result[name] = int(float(value)) else: result[name] = str(value).strip() except Exception: # 转换失败时用默认值兜底，并记录异常状态 result[name] = default return result def parse_to_dataframe(raw_text: str, fields: list) -> pd.DataFrame: json_str = extract_json_block(raw_text) try: raw_data = json.loads(json_str) except json.JSONDecodeError: # JSON解析失败返回空表 return pd.DataFrame(columns=[f["name"] for f in fields]) if isinstance(raw_data, dict): records = [raw_data] elif isinstance(raw_data, list): records = raw_data else: records = [] cleaned_records = [safe_parse_record(r, fields) for r in records] df = pd.DataFrame(cleaned_records, columns=[f["name"] for f in fields]) return df

这里有个细节需要注意：DataFrame(columns=...)能保证即使没有任何记录，返回的DataFrame依然拥有正确的列名，下游拿到空表也能正常处理。

3.3 多文档结果合并成一张表

很多RAG检索场景下，一次查询可能会返回多段文档，模型也可能分别输出多条结果。因此output_parsing01还得支持把多个解析结果合并成一张大表。

实际操作中，我倾向于先把每一条结果解析成DataFrame，再统一用pd.concat合并。这里提醒大家注意索引重置：分开解析的DataFrame索引往往是0到n-1，直接concat会出现重复索引，务必在合并后调用reset_index(drop=True)。

frames = [] for raw_item in model_outputs: df_item = parse_to_dataframe(raw_item, FIELDS) frames.append(df_item) final_df = pd.concat(frames, ignore_index=True)

合并之后，还能顺势做一次空值填充和类型校正，确保整张表符合下游预期。比如把amount列的缺失值统一填0，或者把日期列的None统一显示为空字符串。

4. 实测中踩过的坑和排查思路

4.1 JSON解析失败但不见报错

最常见的问题是模型返回了看起来很像JSON的内容，但包含注释、单引号、尾逗号。Python的标准json库对这些内容会直接报错。我们一开始没有做特殊处理，结果导致部分输出被吞成了空表。

后来我的排查思路是：先记录原始字符串，再逐条看是哪种脏数据。然后针对典型脏格式做修复层：

def repair_json_like(raw_text: str) -> str: # 去掉注释 // 和尾逗号，单引号换成双引号 lines = raw_text.splitlines() cleaned_lines = [] for line in lines: # 去注释，注意别误伤字符串中的内容 if '//' in line and not re.match(r'.*://', line): line = line.split('//')[0] cleaned_lines.append(line) cleaned = "\n".join(cleaned_lines) cleaned = cleaned.replace("'", '"') cleaned = re.sub(r",\s*([}\]])", r"\1", cleaned) return cleaned

简单修复层解决了一大半问题。但说实话，最根本的办法还是把prompt调教好，减少脏数据产生的概率。修复层只能做兜底。

4.2 字段值被模型篡改

模型可能会把字段名“order_id”输出成“orderId”或“订单号”，即使你在字段定义里反复强调。遇到这种情况，我会在解析前增加一层字段名映射。

做法也不复杂：先用所有候选别名去匹配模型返回的key，如果找不到才用默认名。比如：

ALIAS_MAP = { "order_id": ["order_id", "orderId", "订单号"], "order_date": ["order_date", "orderDate", "日期"], "amount": ["amount", "金额"], } def normalize_keys(raw_dict: dict) -> dict: normalized = {} for canonical_name, aliases in ALIAS_MAP.items(): for alias in aliases: if alias in raw_dict: normalized[canonical_name] = raw_dict[alias] break return normalized

这段代码在归拢不同来源的数据时极其有用，强烈建议在解析器中配置别名表。

4.3 类型校验老是失败

日期格式是最容易出问题的字段。模型返回的日期有时候是“2024/06/01”，有时候是“2024年6月1日”。如果只在解析函数里固定用%Y-%m-%d，遇到其他格式就会全部变成默认值。

后来我增加了多格式匹配逻辑：

DATE_FORMATS = [ "%Y-%m-%d", "%Y/%m/%d", "%Y年%m月%d日", "%Y.%m.%d", ] def parse_date(value): for fmt in DATE_FORMATS: try: return datetime.strptime(str(value).strip(), fmt) except ValueError: continue return None

这样能覆盖常见的日期写法和边缘情况，不至于丢掉记录。数值字段也要注意千分位逗号、单位“元”等干扰，解析函数里提前统一清洗。

4.4 大批量解析时性能不够

一开始我们逐条解析、逐条生成DataFrame，处理几百条结果还没什么感觉，但到了上千条就明显变慢。后来做了两处优化：一是把字段转换逻辑集中在一次循环里，不要每列都单独遍历一遍字典；二是尽量以列表推导式生成记录，再用一次性DataFrame构造，避免反复append。

实测下来，解析几千条JSON记录的时间从秒级降到百毫秒级别，效果很明显。做数据处理模块，性能上的良好习惯还是要有的。

5. 从output_parsing01到生产可用的一些心得

这个模块断断续续迭代了几轮，我最大的体会是：输出解析这件事，本质上是在跟概率做对抗。无论prompt设计得多精密，模型总有概率给出你意想不到的格式。所以解析器一定要做到“坏数据不崩、缺字段不炸、类型错了有兜底”。

在工程落地时，我会在解析器里嵌入一个简单的日志记录，记录每次解析成功、失败、字段补充的情况。有了日志，当模型升级或者prompt微调之后，你就能很快发现解析成功率变了多少。否则光是“为什么这段答案变成了空表”就能让人查一晚上。

另外有个小建议：不要试图在一个函数里既做解析又做业务逻辑。像output_parsing01这样保持单一职责，只输出DataFrame，其他环节尽量不要碰。这样将来替换字段定义、调整容错策略时，影响面很小，测试也特别好写。把这些基础打扎实，后面不管接到哪个RAG项目，都能少走一大段弯路。