设备报价整理这件事干过的都知道有多磨人。我之前在某供应链公司帮他们做库存系统对接每天几十封邮件里夹着不同供应商发来的PDF报价单、Excel台账甚至还有直接贴在邮件正文里的纯文本。格式五花八门有的列名叫“设备型号”有的叫“产品编号”有的干脆不叫名字表头是一串部门内部黑话。人工录入的话一个人一天处理两三百条就到头了眼睛盯久了还容易看串行把A设备的年份填到B设备上去。那会儿团队里正好在推 Agent 方向的技术方案我就想着用 LangGraph 做一个结构化数据的自动入库代理。所谓“自动入库”核心就一句话把非结构化的原始文件内容自动转成数据库表里干净的一行行记录。这篇是系列第七篇前面几篇把 State、节点、工具调用、多代理协作都拆得差不多了这一篇就是正儿八经的实战收尾——我把这套流程跑通的完整思路、代码结构、以及前前后后踩过的坑都记在这里。如果你正在做数据入库、单据解析、报表自动化这类活这篇能帮你少走至少两个星期的弯路。1. 整体设计思路为什么用 LangGraph 来做入库代理1.1 入库任务和普通 Agent 任务的本质差别最开始我也想得简单直接拿一个 ReAct 风格的 Agent把“读取文件、提取字段、写入数据库”三个动作全部交给大模型一句 prompt 全部搞定。试了两天就发现不行。问题不在于大模型能不能做而在于它做得不够稳定而且出了问题你根本不知道去哪一层查。入库这类任务和其他问答类 Agent 任务有三个本质差别。第一它必须保证字段级准确性。Chat 场景里模型说错一个年份对话还能继续。但入库场景里一个年份错了一条记录就脏了。更麻烦的是脏数据往往不是当场爆出来而是在后续业务查询、统计报表阶段才浮出来到时候回溯成本极高。第二它需要明确的流程节点。读取文件、字段映射、抽取、清洗、校验、入库每一步的输入输出是固定的。这种流程极其适合用图来建模因为图的每个节点就是一步操作节点的边界就是天然的错误拦截点。第三它需要可干预。纯自动化入库最大的风险是没人把关模型偶尔抽风或者遇到格式极端怪异的文件如果没有人工确认环节错误数据就悄悄进去了。LangGraph 的断点机制和人在环上设计天然就是为了解决这个问题。所以我的结论是结构化数据自动入库不能做成“一把梭”式的单体 Agent而应该是一个状态驱动、节点清晰、带人工确认位的多节点工作流。LangGraph 的 StateGraph 正好是干这个的。1.2 工作流全貌九个节点一条流水线完整的入库流程我拆成了九个节点串起来看是这样load_document - map_fields - extract_entities - clean_values - validate_schema - human_confirmation - write_to_db - log_result另外还有一个独立分支节点 handle_error任何节点抛错都会路由到它。每个节点的职责单一load_document 只负责把文件内容读成纯文本map_fields 只负责把源文件里的字段名映射到目标表字段extract_entities 只做字段抽取clean_values 做格式清洗validate_schema 做完整性校验。节点之间不互相调用唯一传递信息的方式是共享的 State 对象。这里我特别想强调一个设计原则让每个节点成为一个独立的状态转换器。LangGraph 的节点本质上就是接收状态、返回状态更新的函数。错误发生在这个节点就只在这个节点范围内处理不会污染别的节点。这就好比工厂流水线上每一道工序都有质检员哪一道出问题产品就停在那一工位而不是继续往下流。我实测的感受是一旦你开始用这种思维去做 Agent你会发现自己对什么该交给大模型、什么不该交给大模型有了更清晰的判断。比如字段映射这种规则性极强的活根本不需要大模型写一个映射表就够了。只有真正需要语义理解的环节——比如从一段自然语言描述里提取“设备型号”——才值得调用 LLM。1.3 这套方案解决什么场景不解决什么场景这套方案最适合下面这类场景数据源格式混乱、字段名不统一但目标表结构稳定数据量级在每天几百到几千条不需要实时写入业务上允许“先确认后入库”而不是全自动一把梭反过来如果你要处理的是 PB 级数据、需要毫秒级入库、或者目标表结构每天都在变这套方案就不合适了。LangGraph 代理适合的是“业务逻辑复杂、格式多变、需要语义理解”的入库场景不是大规模数据管道场景。后者你应该去用成熟的 ETL 工具。2. 核心细节解析State 设计、节点拆分与循环机制2.1 状态定义入库代理的“共享白板”LangGraph 的核心模型就是共享状态。你可以把 State 理解成一块所有节点都能读写的大白板。每个节点运行完返回对白板的局部更新下一个节点在白板的最新状态上继续操作。这个模型非常直观。我定义的数据入库 State 长这样from typing import TypedDict, List, Dict, Any, Optional class IngestionState(TypedDict): file_path: str # 当前处理的文件路径 source_content: str # 文件读出来的原始文本 field_map: Dict[str, str] # 源字段名 - 目标字段名 的映射 raw_extractions: List[Dict[str, Any]] # LLM 抽取出来的原始记录列表 cleaned_records: List[Dict[str, Any]] # 清洗后的记录列表 validation_errors: List[str] # 校验错误集合 confirmed_records: List[Dict[str, Any]] # 人工确认后的记录 failed_records: List[Dict[str, Any]] # 最终确认失败、拒收的记录 retry_count: int # 当前文件重试次数 status: str # idle / running / awaiting_confirm / done / error这里有几个字段我实际做下来觉得特别重要说细一点。一个是failed_records。很多人做入库代理喜欢“全有或全无”这批文件要么全成功要么全失败。但真实业务不是这样的一个 PDF 里可能 20 条记录里只有 3 条有问题。所以我把“失败记录”单独放在 State 里让这部分走拒收流程剩余 17 条照常入库。这是贴近真实业务需求的一个设计。另一个是retry_count。LLM 抽取不稳定偶尔会抽错或者漏抽。我在校验节点里统计错误数量如果错误率超过阈值就让流程回退到抽取节点带着错误信息重新抽一次。这个就是 LangGraph 里典型的循环场景。图模型允许边从一个节点指回前一个节点这是普通链式调用做不到的。2.2 节点拆分与边界设计九个节点里我自己花最多心思设计的是两个extract_entities 和 validate_schema。这两个节点是整条流水线的灵魂。extract_entities 节点承载着“从自然语言里理解业务实体”的任务。我最开始让这个节点直接读整个 PDF 的全文一次性抽取所有记录。后来发现根本不行——文件一长抽取结果就开始丢字段甚至出现模型“幻觉”出根本不存在的设备。后来改成按段落分批抽取先定位文件里的记录边界比如按表格行、按序号、按分隔符然后每条记录独立调用一次抽取逻辑。这样准确率直接从七成拉到了九成以上。validate_schema 节点是最后一道防线。它做的事情很死板但很关键检查每条记录的目标字段是否齐全字段类型是否符合数据库表定义必填项是否为空。比如“设备状态”这个字段我只允许枚举值在库、已售、维修中、报废。凡是抽取出“状态良好”这种自由文本的统统拦下来进入清洗流程清洗不了就进拒收队列。这里用的是规则校验不靠大模型。规则可以穷举就不要让模型自由发挥。2.3 条件边与循环让流程具备纠错能力LangGraph 里最让我觉得“值回票价”的功能就是条件边。简单说你可以在一个节点执行完后根据当前 State 的内容动态决定下一个节点是谁。我在 validate_schema 和 extract_entities 之间加了一条回边如果 validation_errors 里的错误数量超过阈值并且 retry_count 小于 2就回到抽取节点重新抽取。同时我在 State 里把上次的错误信息拼接进抽取节点的 prompt 里相当于告诉模型“你上一次把年份抽成了型号这次注意”。这个逻辑写成代码就是def should_retry(state: IngestionState) - str: if state[retry_count] 2 and len(state[validation_errors]) 3: return extract_entities return clean_values整个过程最多重试两次第三次仍然失败直接进 handle_error 节点转人工处理。这样避免了两头风险一是不会因为模型反复失败陷入死循环二是不会因为只试一次就放弃本来可以救回来的数据。3. 实操过程从零搭建一个可用的入库代理3.1 环境准备与核心依赖我用的环境是 Python 3.11核心依赖其实只有几个pip install langgraph langchain langchain-openai pandas openpyxl如果你对接的数据库是 MySQL需要额外装 pymysql用 SQLite 的话 Python 自带 sqlite3 就够了不需要额外装东西。我这里做了一个取舍说明为什么用 SQLite 做演示因为这套流程的核心价值在“抽取和清洗环节”数据库本身只是一个落点。SQLite 零配置、单文件、方便拆解最适合做流程验证。真正上生产把写入节点里的数据库连接换成 MySQL 或 PostgreSQL 就行其他代码几乎不用动。LLM 这一层我建议选支持结构化输出约束的模型效果好很多。具体来说就是能让模型严格返回 JSON 格式并且字段类型受约束的那种。我用的是 Qwen 系列和 GLM 系列都跑过表现稳定。关键点不在于具体用的是哪家模型而在于你的 prompt 和解析逻辑要足够稳。后面我会专门讲 prompt 怎么设计。3.2 确定性映射在前LLM 抽取在后我在 1.2 里提了一句“能写规则就不要用大模型”现在具体展开说。field_mapping 这一步我的做法是先写一个字典做硬映射。比如FIELD_ALIASES { 设备型号: model_no, 型号: model_no, 产品编号: model_no, 设备名称: device_name, 名称: device_name, 出厂年份: year, 年份: year, 报价金额: price, 价格: price, 供应商: supplier, 购买日期: purchase_date, 设备状态: status, 备注: remark }凡是能在映射表里直接命中的直接转换不走 LLM。只有源文件里的字段名不在映射表里或者干脆是自由文本格式的行记录才交给 LLM 抽取。这样做的原因很朴素映射规则是确定的、可解释的、免费的而 LLM 调用是有成本、有延迟、有随机性的。3.3 抽取节点的 Prompt 与输出解析extract_entities 节点内部的逻辑是接收源文件片段和当前字段映射表构造抽取 prompt调用 LLM要求输出严格 JSON解析 JSON塞进 raw_extractions核心 prompt 我调了好几版最终稳定下来的长这样EXTRACT_PROMPT 你是一个数据抽取引擎。请从下面给定的文本片段中抽取设备报价记录。 要求 1. 输出格式必须是 JSON 数组数组每个元素是一条记录。 2. 每条记录必须包含以下字段model_no, device_name, year, price, supplier, purchase_date, status。 3. 如果某个字段在原文中不存在用 null 表示不要编造。 4. 如果一条文本里包含多条设备记录必须全部抽出来一条不能丢。 5. 如果有无法判断的记录保留所有原始文本片段在 remark 字段里不要丢弃。 字段说明 - model_no: 设备型号通常是字母数字组合 - device_name: 设备名称或类别 - year: 出厂年份四位数 - price: 报价金额只保留数字单位是元 - supplier: 供应商名称 - purchase_date: 购买日期格式YYYY-MM-DD - status: 设备状态只允许取值为在库、已售、维修中、报废 文本片段 {chunk} 这里我踩过一个很深的坑最开始我让模型输出“包含一条记录的 JSON 对象”而不是“JSON 数组”。结果遇到一个文件里有 5 条记录时模型只抽了第一条就收了。后来我改成强制 JSON 数组并且在 prompt 里明确写“必须抽完一条不能丢”总算把漏抽率压下来了。还有一个细节是 remark 字段。我让模型把无法判断的原始文本片段原样保留在 remark 里这样即使某个字段抽不出来原始数据也不至于丢失。人工复审时可以直接看 remark不用回头再翻原始文件。这个设计救了我好多次。解析输出的时候要注意容错。我用的是 langchain 的输出解析器再加上一层手动兜底from langchain_core.output_parsers import JsonOutputParser parser JsonOutputParser() try: parsed parser.invoke(llm_response) except Exception: # 兜底从原始文本里截取 [...] start llm_response.find([) end llm_response.rfind(]) 1 parsed json.loads(llm_response[start:end])模型输出偶尔会带一点解释性文字比如“好的下面是抽取结果”。质量好的模型很少这样但为了稳兜底逻辑必须写。3.4 清洗、校验与去重抽取完成后清洗节点做这几件事年份字段统一转成四位数。有的源文件写“98年”转成“1998”。价格字段去掉货币符号、千分位逗号统一转成浮点数。日期字段统一转成 YYYY-MM-DD。状态字段做枚举映射比如“已出售”归一成“已售”“在用”归一成“在库”。清洗完进 validate_schema规则如下model_no 和 device_name 不允许为空year 必须是 1900 到当前年份之间的整数price 必须大于 0status 必须是四个枚举值之一然后还有一个非常重要的去重逻辑。同一家供应商可能一周内发两次报价表第二次是修订版如果直接入库就会出现重复记录。我的做法是在目标表里建唯一索引字段组合是 (model_no, supplier, year, price)重复的记录在入库前直接挡掉。这部分的代码我写成了一个小函数集统一挂在 clean_values 节点里def clean_record(record: dict) - dict: record[year] normalize_year(record.get(year)) record[price] normalize_price(record.get(price)) record[purchase_date] normalize_date(record.get(purchase_date)) record[status] normalize_status(record.get(status)) return record def validate_record(record: dict) - List[str]: errors [] if not record.get(model_no): errors.append(model_no 缺失) if not record.get(device_name): errors.append(device_name 缺失) if not is_valid_year(record.get(year)): errors.append(year 不合法) if not is_valid_price(record.get(price)): errors.append(price 不合法) if record.get(status) not in VALID_STATUS: errors.append(status 枚举值不合法) return errors3.5 人工确认与入库落库清洗校验通过后我设计了一个 human_confirmation 节点。这个节点在 LangGraph 里用到一个特性叫 interrupt简单说就是让图在某个节点暂停等待外部输入再继续。实际跑的时候这批待确认记录是攒下来在系统里生成一个待办任务人工在网页上扫一眼有问题就改没问题就批量点“确认”。有人可能会说这不是增加工作量吗操过系统的人都懂宁可加一道轻量的人工确认也不要事后花三倍时间洗数据。我的经验是人工只需要关注 flagged 的记录也就是校验有警告但不至于拒收的那些正常情况下几十条记录几十秒就确认完了不算负担。确认后的记录进入 write_to_db这一步就完全是常规数据库操作了。先检查唯一索引不重复的 insert重复的进 failed_records。最后 log_result 节点把处理结果记录到日志表整个流程结束。4. 完整代码骨架一个最小可跑的 LangGraph 入库图下面给一个可以实际跑起来的最小骨架。这个骨架我把数据库简化成了 SQLiteLLM 调用做了接口抽象方便你替换成自己的模型。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from typing import Literal # 节点函数定义 def load_document(state: IngestionState) - dict: with open(state[file_path], r, encodingutf-8) as f: content f.read() return {source_content: content, status: running} def map_fields(state: IngestionState) - dict: # 硬映射 未知字段标记 mapping apply_field_aliases(state[source_content]) return {field_map: mapping} def extract_entities(state: IngestionState) - dict: chunks split_into_chunks(state[source_content]) records [] for chunk in chunks: prompt EXTRACT_PROMPT.format(chunkchunk) llm_result llm_with_structured_output(prompt) parsed safe_parse_json(llm_result) records.extend(parsed) return {raw_extractions: records} def clean_values(state: IngestionState) - dict: cleaned [clean_record(r) for r in state[raw_extractions]] return {cleaned_records: cleaned} def validate_schema(state: IngestionState) - dict: errors [] valid_records [] for rec in state[cleaned_records]: rec_errors validate_record(rec) if rec_errors: errors.append({record: rec, errors: rec_errors}) else: valid_records.append(rec) return { validation_errors: errors, cleaned_records: valid_records, retry_count: state[retry_count] 1, } def handle_error(state: IngestionState) - dict: # 拒收记录失败原因 for err_item in state[validation_errors]: state[failed_records].append(err_item) return {failed_records: state[failed_records], status: error} def write_to_db(state: IngestionState) - dict: # 简化版插入 SQLite唯一索引冲突的记录进 failed import sqlite3 conn sqlite3.connect(inventory.db) cursor conn.cursor() inserted 0 for rec in state[confirmed_records] if state.get(confirmed_records) else state[cleaned_records]: try: cursor.execute( INSERT INTO devices (model_no, device_name, year, price, supplier, purchase_date, status, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?), (rec[model_no], rec[device_name], rec[year], rec[price], rec.get(supplier), rec.get(purchase_date), rec[status], rec.get(remark)) ) inserted 1 except sqlite3.IntegrityError: state[failed_records].append(rec) conn.commit() conn.close() return {failed_records: state[failed_records], status: done} # 条件路由 def should_retry(state: IngestionState) - Literal[extract_entities, clean_values]: if state[retry_count] 2 and len(state[validation_errors]) 3: return extract_entities return clean_values def route_after_validate(state: IngestionState) - str: if state[validation_errors] and state[retry_count] 2: return handle_error return write_to_db # 构建图 graph StateGraph(IngestionState) graph.add_node(load_document, load_document) graph.add_node(map_fields, map_fields) graph.add_node(extract_entities, extract_entities) graph.add_node(clean_values, clean_values) graph.add_node(validate_schema, validate_schema) graph.add_node(handle_error, handle_error) graph.add_node(write_to_db, write_to_db) graph.set_entry_point(load_document) graph.add_edge(load_document, map_fields) graph.add_edge(map_fields, extract_entities) graph.add_edge(extract_entities, clean_values) graph.add_conditional_edges(clean_values, lambda s: validate_schema) graph.add_conditional_edges( validate_schema, should_retry, {extract_entities: extract_entities, clean_values: write_to_db} ) graph.add_edge(handle_error, END) graph.add_edge(write_to_db, END) app graph.compile(checkpointerMemorySaver())这段代码如果你之前跟着系列文章跑过前面的例子读起来应该很顺。核心就三件事定义状态类型写节点函数把节点连成图。我觉得 LangGraph 最好的体验就是你不需要发明一个新的编程范式节点就是普通函数图就是函数之间的路由关系。5. 常见问题与排查技巧实录这一部分我直接整理成速查表全部来自我实际跑流程时踩过的坑不是网上抄来的。问题现象根本原因解决方案同一个文件每次跑结果不一样LLM 输出的天然随机性prompt 里开启温度归零对关键字段用“限定枚举值 JSON 严格模式”约束模型漏抽记录只返回部分条目一次调用塞了太多文本进 prompt按记录边界切块每条独立抽取prompt 明确“一条不能丢”日期字段格式五花八门源文件来自不同渠道清洗节点做统一归一化不接受模型自由输出同一设备重复入库供应商发了修订版文件目标表加唯一索引 入库前查重校验总是报“字段缺失”部分字段在源文件里确实不存在必填项和可空项要严格区分可空字段允许为 null错误记录把整批流程卡死我没有设计错误分支给每个危险操作加“出错则进 failed_records剩余继续”的逻辑人工确认环节被跳过忘了挂 checkpointerLangGraph 的人为中断必须配合 checkpointer 使用否则状态无法持久化这里我特别展开说两个。第一个是“温度和随机性”的问题。我在抽取节点里一开始把 temperature 设成 0.3想着给模型一点创造性空间能让它更好地理解上下文。后来发现完全没必要抽取是确定性任务不是创意写作。把 temperature 直接设成 0 之后同一条文本抽出来的结果稳定多了。这个改动虽小收益非常明显。第二个是“切块大小”。我做过一组对比实验一段 2000 字的设备清单一次性全塞进去抽20 条里漏了 4 条还有 2 条年份抽错。把同样内容按“每个供应商块”切成 4 块分别抽20 条全部命中错误 0。道理很简单模型在长上下文末尾容易注意力涣散这就是大家常说的“lost in the middle”现象。宁可多调几次模型也要把每次输入的长度控制住。我自己的经验阈值是抽取类型的 prompt 输入控制在 1500 字以内效果最稳。第三个坑是关于“remark 字段”的使用。这个我在 3.3 里提过但值得再强调一次。你可以把 remark 理解成“安全网”。凡是模型拿不准的信息都让它原样留在 remark 里不要强迫它猜一个填进去。入库后如果业务发现某条数据有疑点DBA 直接搜 remark 就能定位到原始文本。这个习惯帮我避免了好几次“死无对证”的尴尬。6. 实践经验总结与后续扩展说实话这个入库代理做出来之后最大的感受不是“代码多复杂”而是“流程设计比模型调用更重要”。LangGraph 真正解决的问题是把一个不可控的 LLM 调用过程放进了一个可控的工程框架里。你可以对每一步做校验、做重试、做人工确认出现问题知道去哪里查。几个我后来在类似项目里继续沿用的经验凡是确定性逻辑坚决不调用 LLM省成本也省延迟。凡是 LLM 输出必须过一层严格 schema 校验绝不让脏数据直接入库。人工确认位不是“兜底”是整个流程的正式一环建议 UI 上做待办列表而不是塞在聊天窗口里。所有失败记录一定要保留原始证据原始文本片段否则后续排查就是大海捞针。这个项目后续还可以继续扩展的方向有三个。一是加入多模态能力直接识别 PDF 里扫描图片中的报价表格这需要接 OCR 组件二是把抽取结果接入向量库做个语义检索方便业务侧直接问“最近有没有单价低于 5000 的设备”三是做调度层定时去邮件附件、共享目录里拉文件全自动触发入库流程。每一步都在现有图结构上加节点就行不需要推翻重来。最后分享一个小技巧调试 LangGraph 流程时别光看最终结果把每个节点的中间状态打印出来看一遍。我自己写了个简单的 State 日志函数每次运行完把 State 的关键字段落盘成 JSON。一开始觉得麻烦后来全靠它定位问题。做 Agent 项目过程可见性和结果正确性一样重要。