1. 为什么大模型不能只当聊天工具用

1.1 一个被忽视的现实：数据管道里的“最后一公里”问题

干了这么多年数据工程，我发现一个特别有意思的现象：几乎每个团队都在聊大模型，但绝大多数人只把它当成一个更聪明的聊天窗口。写邮件、查资料、生成代码片段，这些场景确实好用，但真正让数据团队头疼的问题——非结构化数据进数仓——却很少有人认真讨论。

什么叫非结构化数据进数仓？简单说，你手头有一堆合同PDF、客服对话记录、工单文本、商品评论、会议纪要，这些东西里面藏着巨大的业务价值，但它们没法像数据库表那样直接SELECT出来。传统做法是人工打标签、写正则表达式、做关键词提取，费时费力不说，准确率还堪忧。一个典型的场景：客服团队每天产生上万条对话记录，业务方想知道“客户最常抱怨的三个问题是什么”，以前只能抽样看几百条，现在有了大模型，理论上可以全量分析。但问题来了——分析完之后呢？结果怎么落表？怎么和现有的数仓维度关联？怎么保证每天增量更新？

这就是我说的“最后一公里”问题。大模型能理解文本，但它不理解你的数据管道。它不知道你的数仓里有一张dim_customer表，不知道你的分区键是dt，不知道你的ETL调度是每天凌晨两点跑。你要做的，是在大模型和数据仓库之间架一座桥，让非结构化数据经过大模型处理后，变成结构化的、可查询的、可关联的字段，真正进入你的数据资产体系。

1.2 非结构化数据进数仓的核心挑战

先把这个问题的难点拆开来看，不然后面讲方案的时候容易飘。

第一个挑战是Schema设计。非结构化数据的“结构”不是预先定义的，而是你根据业务需求“问”出来的。比如一段客服对话，你可以提取“客户情绪”“问题分类”“涉及产品”“解决状态”等字段，但这些字段的粒度、枚举值、是否可空，都需要你提前想清楚。更麻烦的是，大模型的输出是自然语言，它可能今天返回“物流延迟”，明天返回“配送慢”，后天返回“发货太慢”，这三个词在业务上是一个意思，但在数仓里就是三个不同的值。你需要一套标准化机制。

第二个挑战是数据量。大模型推理是有成本的，不管是调用API还是本地部署，吞吐量都远低于传统的正则匹配。如果你有每天千万级的文本要处理，全量跑大模型不现实。你需要做增量识别、优先级排序、批量合并。

第三个挑战是数据质量。大模型会“幻觉”，会漏提取，会过度解读。你让它从合同里提取“签约金额”，它可能把“违约金”也当成签约金额。你让它判断情感倾向，它可能把反讽当成正面。这些错误如果直接进数仓，下游的报表和决策就全歪了。你需要一套校验和回退机制。

第四个挑战是工程集成。大模型处理只是管道中的一环，前面要有数据采集和清洗，后面要有质量校验和入库，中间还要考虑失败重试、限流、缓存、监控。这些工程细节，才是决定方案能不能上生产的关键。

1.3 谁适合参考这套方案

如果你是一个数据工程师，正在被业务方追着要“文本数据的结构化报表”，这套方案能帮你理清思路。如果你是一个数据产品经理，想知道大模型在数据管道里到底能干什么、不能干什么，这篇文章会给你一个务实的答案。如果你是一个刚接触大模型应用的开发者，想了解怎么把模型输出变成可靠的数据资产，下面的内容会涉及具体的Schema设计、Prompt工程、质量校验和调度策略。

我不打算讲太多理论，重点放在“怎么做”和“踩过哪些坑”上。所有代码和配置都是示意性的，你可以根据自己团队的技术栈调整。

2. 整体架构设计：从文本到数仓的完整链路

2.1 分层设计思路

把大模型嵌入数据管道，最忌讳的就是“一把梭”——把所有文本丢给模型，让它直接输出JSON，然后直接入库。这种做法的失败率极高，因为中间没有任何缓冲和校验环节。

我推荐的架构是分层的，每一层只做一件事，层与层之间通过标准化的中间状态衔接。具体来说，分为五层：

• 采集层：负责从各种数据源（数据库、日志文件、消息队列、对象存储）拉取原始文本数据，做初步的清洗和去重。
• 预处理层：对文本做分块、截断、脱敏、语言识别，确保输入大模型的文本是干净的、长度可控的。
• 推理层：调用大模型进行信息抽取、分类、摘要、情感分析等任务，输出结构化的候选结果。
• 校验层：对模型输出做格式校验、枚举值校验、一致性校验，过滤掉明显错误的结果。
• 入库层：将校验通过的数据写入数仓的ODS或DWD层，并建立与现有维度表的关联。

这个分层的好处是，每一层都可以独立测试、独立扩展、独立替换。比如你今天用A模型，明天想换B模型，只需要改推理层，其他层不动。再比如你发现某类文本的提取准确率低，可以在预处理层加规则过滤，而不需要动模型。

2.2 为什么选择“大模型+规则”的混合方案

纯大模型方案听起来很美好，但实际落地时会遇到几个硬伤。第一是成本，全量调用大模型的费用可能比人工标注还贵。第二是稳定性，大模型的输出有随机性，同样的输入可能得到不同的输出，这对需要可重复的数据管道来说是致命的。第三是可解释性，当业务方问“为什么这条记录被分类为投诉”，你很难用大模型的注意力权重去解释。

所以我的建议是混合方案：用规则做粗筛和兜底，用大模型做精细理解和语义归一。具体来说：

• 对于格式固定的文本（比如日志、模板化邮件），优先用正则或关键词匹配，成本低、速度快、结果稳定。
• 对于格式不固定但语义简单的文本（比如短评论），可以用小模型或大模型的轻量版本做分类。
• 对于语义复杂、需要深度理解的文本（比如长合同、多轮对话），才用大模型做信息抽取。
• 大模型输出后，再用规则做一轮校验和归一化，比如把“配送慢”“物流延迟”“发货太慢”统一映射到“物流问题”这个枚举值。

这种混合方案的核心思想是：让大模型做它擅长的事（语义理解），让规则做它擅长的事（格式校验和标准化）。两者结合，才能在准确率、成本和稳定性之间找到平衡。

2.3 数据流与状态管理

整个管道的数据流可以用一个简单的状态机来描述。每条原始文本记录从进入管道开始，会经历以下状态：

这个状态机需要持久化存储，通常用一张元数据表来管理。每次状态变更都要记录时间戳、操作人（或任务ID）、变更原因。这样做的好处是，当管道出现问题时，你可以快速定位是哪一批数据、在哪个环节卡住了。

注意：状态管理表本身也要考虑分区和清理策略。如果每天处理千万级记录，状态表会迅速膨胀。建议按天分区，保留最近30天的详细状态，更早的只保留聚合统计。

3. 核心细节解析：Schema设计、Prompt工程与质量校验

3.1 如何设计一个“可落地”的抽取Schema

Schema设计是整个方案的地基。设计得不好，后面怎么调Prompt都救不回来。

我的经验是，Schema设计要遵循三个原则：业务驱动、枚举可控、粒度适中。

业务驱动的意思是，你抽取的字段必须是业务方真正需要的。不要因为“大模型能提取”就去提取。比如你从客服对话里提取“客户提到的竞品名称”，这个字段业务方可能根本不用，那它就是噪音。正确的做法是先和业务方对齐：你们到底想分析什么？是客户满意度、问题分布、还是产品改进点？根据分析目标反推需要的字段。

枚举可控的意思是，对于分类字段，枚举值必须是有限且明确的。你不能让大模型自由发挥，否则它会给你返回各种同义词。比如“问题分类”这个字段，你应该提前定义好枚举值：物流问题、产品质量、售后服务、支付问题、其他。然后在Prompt里明确告诉模型：只能从这五个值里选一个。

粒度适中的意思是，字段的粒度要匹配业务的分析需求。太粗了没信息量，太细了没法聚合。比如“客户情绪”这个字段，用正面/中性/负面三分类就够了，不需要搞成非常满意/满意/一般/不满意/非常不满意五分类，因为后者的标注一致性很难保证，大模型也容易分错。

下面是一个客服对话抽取的Schema示例：

{ "conversation_id": "string, 对话唯一标识", "customer_intent": "enum(咨询, 投诉, 建议, 其他)", "issue_category": "enum(物流问题, 产品质量, 售后服务, 支付问题, 其他)", "sentiment": "enum(正面, 中性, 负面)", "involved_product": "string, 产品名称或ID，无则填null", "resolution_status": "enum(已解决, 未解决, 待跟进)", "summary": "string, 50字以内的对话摘要" }

这个Schema里，conversation_id是主键，用于关联原始数据；customer_intent和issue_category是分类字段，枚举值明确；sentiment是三分类；involved_product是可选字段；resolution_status用于跟踪处理进度；summary是自由文本，用于快速浏览。

实操心得：Schema不要一次设计太多字段。我见过一个团队设计了30多个字段，结果Prompt变得极其复杂，模型准确率直线下降。建议从5-8个核心字段开始，跑通之后再逐步扩展。

3.2 Prompt工程：让大模型稳定输出结构化结果

Prompt工程是连接自然语言和结构化数据的桥梁。一个好的Prompt应该做到：角色明确、任务清晰、格式固定、示例充分。

角色明确：告诉模型它是什么角色。比如“你是一个客服对话分析助手，负责从对话中提取结构化信息。”这比直接说“提取以下字段”效果要好，因为模型会根据角色调整它的“注意力”。

任务清晰：把任务拆成具体的步骤。不要一次性让模型做太多事。比如先让它判断对话类型，再提取问题分类，最后生成摘要。分步做的好处是，每一步的准确率都更高，而且出错时容易定位。

格式固定：明确告诉模型输出格式。最好用JSON，因为JSON有严格的语法，便于程序解析。在Prompt里给出JSON的模板，并强调“只输出JSON，不要输出其他内容”。

示例充分：给2-3个完整的输入输出示例。示例要覆盖典型场景和边界情况。比如一个正面情绪的对话、一个负面情绪的对话、一个无法分类的对话。示例的质量直接决定模型的输出质量。

下面是一个实际的Prompt模板：

你是一个客服对话分析助手。请从以下对话中提取结构化信息。 对话内容： {conversation_text} 请按照以下JSON格式输出，不要输出任何其他内容： { "customer_intent": "咨询|投诉|建议|其他", "issue_category": "物流问题|产品质量|售后服务|支付问题|其他", "sentiment": "正面|中性|负面", "involved_product": "产品名称或null", "resolution_status": "已解决|未解决|待跟进", "summary": "50字以内的摘要" } 示例1： 对话内容：客户问“我的订单什么时候到？”客服回复“预计明天送达。” 输出：{"customer_intent": "咨询", "issue_category": "物流问题", "sentiment": "中性", "involved_product": null, "resolution_status": "已解决", "summary": "客户咨询订单配送时间，客服告知预计明天送达。"} 示例2： 对话内容：客户说“你们的产品质量太差了，用了两天就坏了！”客服回复“非常抱歉，我们可以为您换货。” 输出：{"customer_intent": "投诉", "issue_category": "产品质量", "sentiment": "负面", "involved_product": null, "resolution_status": "已解决", "summary": "客户投诉产品质量问题，客服提供换货方案。"}

这个Prompt的关键点：枚举值用|分隔，明确可选范围；示例覆盖了咨询和投诉两种典型场景；要求只输出JSON，避免模型加解释性文字。

踩过的坑：早期我在Prompt里写“请尽量提取”，结果模型经常返回“无法确定”或“信息不足”。后来改成“必须从枚举值中选择一个，如果无法判断则选‘其他’”，输出就稳定多了。不要让模型有“不确定”的选项，要给它一个兜底的枚举值。

3.3 质量校验：把错误挡在数仓门外

大模型的输出不能直接信。我见过太多案例，模型把“不推荐”理解成“推荐”，把“退款成功”提取成“退款失败”。如果不做校验，这些错误数据进了数仓，下游的报表就是灾难。

校验分三层：格式校验、枚举校验、一致性校验。

格式校验是最基础的。检查模型输出是不是合法JSON，字段是否齐全，类型是否正确。比如summary必须是字符串，sentiment必须是三个枚举值之一。这一步可以用JSON Schema来做，Python的jsonschema库就能搞定。

枚举校验是检查分类字段的值是否在预定义范围内。如果模型返回了“物流延迟”而不是“物流问题”，就判定为校验失败。这时候有两种处理方式：一是直接拒绝，让这条记录进入人工复核队列；二是做映射，把“物流延迟”归一化到“物流问题”。我建议对高频出现的非标准值做映射，对低频的做拒绝。

一致性校验是检查字段之间的逻辑关系。比如resolution_status是“已解决”，但summary里写的是“客户表示问题仍未解决”，这就矛盾了。再比如sentiment是“正面”，但customer_intent是“投诉”，这也可疑。一致性校验可以用规则引擎来做，也可以用另一个大模型来做“裁判”。

下面是一个校验规则的示例：

import json from jsonschema import validate, ValidationError schema = { "type": "object", "properties": { "customer_intent": {"type": "string", "enum": ["咨询", "投诉", "建议", "其他"]}, "issue_category": {"type": "string", "enum": ["物流问题", "产品质量", "售后服务", "支付问题", "其他"]}, "sentiment": {"type": "string", "enum": ["正面", "中性", "负面"]}, "involved_product": {"type": ["string", "null"]}, "resolution_status": {"type": "string", "enum": ["已解决", "未解决", "待跟进"]}, "summary": {"type": "string", "maxLength": 100} }, "required": ["customer_intent", "issue_category", "sentiment", "resolution_status", "summary"] } def validate_output(output_str): try: data = json.loads(output_str) validate(instance=data, schema=schema) # 一致性校验 if data["customer_intent"] == "投诉" and data["sentiment"] == "正面": return False, "投诉对话的情感不应为正面" if data["resolution_status"] == "已解决" and "未解决" in data["summary"]: return False, "解决状态与摘要矛盾" return True, data except (json.JSONDecodeError, ValidationError) as e: return False, str(e)

这个校验函数会返回校验结果和错误原因。校验失败的记录进入rejected状态，可以后续人工复核，也可以直接丢弃（取决于业务对准确率的要求）。

实操心得：校验规则不要写得太死。我一开始要求summary必须包含动词，结果很多合理的摘要被拒了。后来改成只检查长度和是否为空，通过率就正常了。校验的目的是挡住明显错误，不是追求完美。

4. 实操过程：从零搭建一条文本处理管道

4.1 环境准备与工具选型

先说一下我用的技术栈，你可以根据自己团队的情况调整。核心原则是：用熟悉的工具，不要为了追新而引入不必要的复杂度。

• 数据采集：用Python脚本从数据库和对象存储拉取数据，写入一个中间表。如果数据量大，可以用Spark或Flink。
• 预处理：用Python的re和jieba做文本清洗和分块。脱敏用正则替换手机号、身份证号等敏感信息。
• 大模型调用：用HTTP API调用大模型服务，封装一个带重试和限流的客户端。如果本地部署，可以用vLLM或TGI做推理服务。
• 校验：用jsonschema做格式校验，用自定义规则做一致性校验。
• 入库：用SQLAlchemy或PySpark写入数仓，支持分区覆盖和增量追加。
• 调度：用Airflow或Dagster做任务编排，每天凌晨跑增量。

这个技术栈没有用什么高大上的东西，但胜在稳定、可维护、容易排查问题。

4.2 数据采集与预处理的关键步骤

采集层要做的事情不多，但有几个细节容易忽略。

第一是去重。同一个对话可能被多次采集，你需要一个唯一标识来做去重。通常用业务ID（如conversation_id）加上时间戳来做幂等写入。

第二是增量识别。不要每次都全量跑。用一个last_processed_time字段记录上次处理到的时间点，每次只拉取这个时间点之后的新数据。如果数据源没有时间戳，可以用自增ID或文件修改时间。

第三是文本清洗。原始文本里可能有HTML标签、特殊字符、乱码。用正则做一轮清洗，把<br>换成换行，把连续空格合并，把不可见字符去掉。

预处理层的核心是分块和截断。大模型的上下文长度有限，长文本必须分块。分块的策略取决于任务：如果是做摘要，按段落分块；如果是做信息抽取，按语义单元分块（比如一轮对话）。分块之后，如果某块还是太长，就截断，但要保留开头和结尾，因为关键信息通常在两端。

import re def clean_text(text): # 去除HTML标签 text = re.sub(r'<[^>]+>', '', text) # 合并连续空白字符 text = re.sub(r'\s+', ' ', text) # 去除不可见字符 text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text) return text.strip() def split_chunks(text, max_length=2000): # 按段落分块 paragraphs = text.split('\n') chunks = [] current = "" for p in paragraphs: if len(current) + len(p) > max_length: if current: chunks.append(current) current = p else: current += "\n" + p if current: chunks.append(current) return chunks

注意：脱敏要在分块之前做。如果先分块再脱敏，可能会把手机号截断，导致脱敏失败。脱敏之后要记录脱敏规则，方便后续审计。

4.3 大模型推理的批量处理与限流策略

推理层是成本最高的环节，也是最容易出问题的环节。核心要解决三个问题：批量处理、失败重试、限流控制。

批量处理：不要一条一条调用API，那样网络开销太大。把多条记录合并成一个请求，让模型一次性处理。但要注意，合并后的文本不能超过模型的上下文限制。通常一次处理5-10条比较合适。

失败重试：API调用可能因为网络抖动、服务限流、模型超时而失败。需要实现指数退避的重试策略。第一次失败等1秒，第二次等2秒，第三次等4秒，最多重试3次。如果还失败，记录到失败队列，后续人工处理。

限流控制：大模型服务通常有QPS限制。你需要一个令牌桶或漏桶算法来控制请求速率。如果用的是云服务，注意看它的配额说明；如果是本地部署，根据GPU显存和推理速度来估算。

import time import requests from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) def call_llm(prompt, api_key, endpoint): headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} payload = { "model": "your-model-name", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, # 低温度，保证输出稳定 "max_tokens": 500 } response = requests.post(endpoint, headers=headers, json=payload, timeout=30) response.raise_for_status() return response.json()["choices"][0]["message"]["content"]

这里temperature设为0.1，是为了让输出尽量确定。如果设为0.7或更高，同样的输入可能得到不同的输出，这对数据管道来说是灾难。

实操心得：我试过用temperature=0，但有些模型服务不支持，会报错。0.1是一个比较安全的折中值。另外，max_tokens要设得比预期输出长一些，避免JSON被截断。

4.4 校验与入库的完整实现

校验层和入库层是最后一道关卡。校验通过的记录写入数仓的ODS层，校验失败的记录写入一个rejected表，供人工复核。

入库时要注意几点：

第一是分区。按天分区是最常见的做法。分区键用处理日期，而不是数据产生日期。这样方便重跑和回溯。

第二是幂等。同一条记录可能被多次处理（比如重跑任务），入库时要保证不会重复。可以用INSERT OVERWRITE覆盖当天分区，或者用MERGE做upsert。

第三是关联维度。如果数仓里有dim_customer、dim_product等维度表，入库时要尽量关联。比如从对话中提取的involved_product，可以通过产品名称映射到product_id。关联不上的填-1或null。

def load_to_warehouse(validated_records, partition_date): # 写入ODS层 df = spark.createDataFrame(validated_records) df.write.mode("overwrite").partitionBy("dt").parquet(f"/warehouse/ods/conversation_analysis/dt={partition_date}") # 关联维度表 dim_product = spark.table("dim_product") df_with_product = df.join(dim_product, df.involved_product == dim_product.product_name, "left") \ .select(df["*"], dim_product["product_id"]) df_with_product.write.mode("overwrite").partitionBy("dt").parquet(f"/warehouse/dwd/conversation_analysis/dt={partition_date}")

这个流程跑通之后，业务方就可以在数仓里直接查询了。比如“过去一周物流问题的投诉量趋势”：

SELECT dt, COUNT(*) as complaint_count FROM dwd.conversation_analysis WHERE issue_category = '物流问题' AND customer_intent = '投诉' AND dt >= date_sub(current_date, 7) GROUP BY dt ORDER BY dt;

5. 常见问题与排查技巧实录

5.1 模型输出不稳定怎么办

这是最常见的问题。同样的输入，模型今天返回“物流问题”，明天返回“配送问题”。原因通常是temperature设得太高，或者Prompt里的枚举值不够明确。

解决方案：

• 把temperature降到0.1以下。
• 在Prompt里用|明确列出所有可选值，并强调“只能从以下值中选择”。
• 在Prompt里加一句“如果无法判断，请选择‘其他’”，给模型一个兜底选项。
• 如果还是不稳定，可以在校验层做映射，把高频的非标准值归一化到标准值。

5.2 长文本被截断导致信息丢失

大模型的上下文长度有限，长文本必须截断。但截断可能丢掉关键信息。比如一份合同，金额在开头，违约条款在结尾，中间截断了就提取不到。

解决方案：

• 用滑动窗口分块，块与块之间保留重叠部分。
• 对每块分别推理，然后合并结果。合并时用规则做去重和冲突解决。
• 如果任务对完整性要求高，可以用“摘要+抽取”两阶段：先让模型生成全文摘要，再从摘要中抽取字段。

5.3 成本超预算怎么优化

大模型调用成本可能很高，尤其是全量处理的时候。

优化策略：

• 用规则做粗筛，只把真正需要语义理解的文本送给大模型。
• 用更小的模型做分类任务，大模型只做复杂抽取。
• 缓存推理结果，同样的输入不要重复调用。
• 批量合并请求，减少网络开销。
• 如果本地部署，用量化模型降低显存占用。

5.4 常见问题速查表

踩过的坑：有一次任务重跑，因为入库用的是append模式，导致数据翻倍。后来改成overwrite当天分区，问题解决。任何数据管道都要考虑重跑场景，幂等是底线。

5.5 独家避坑技巧

技巧一：先跑小样本，再全量。不要一上来就全量跑。先拿100条数据测试，看Prompt效果、校验通过率、入库结果。确认没问题再扩大规模。

技巧二：保留原始输出。模型返回的原始JSON要存一份，不要只存校验后的结果。这样当校验规则调整时，可以重新校验，不需要重新调用模型。

技巧三：监控校验通过率。如果通过率突然下降，说明模型服务或Prompt出了问题。设置告警，通过率低于90%就通知。

技巧四：人工复核队列要可控。校验失败的记录不要无限堆积。设置一个上限，超过上限就自动丢弃或降级处理。否则人工复核会成为瓶颈。

技巧五：版本化管理Prompt。Prompt的每次修改都要记录版本号和修改原因。不同版本的Prompt可能产生不同的输出分布，版本化管理方便回溯和对比。

这套方案我在几个项目里跑过，从每天几千条到几十万条的量级都验证过。核心体会是：大模型不是银弹，它只是管道里的一个组件。真正决定成败的，是Schema设计、校验规则和工程细节。把这三样做好，非结构化数据进数仓就不再是难题。