企业官网建设流程全解析

设备报价整理这件事，干过的都知道有多磨人。我之前在某供应链公司帮他们做库存系统对接，每天几十封邮件里夹着不同供应商发来的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 节点内部的逻辑是：

1. 接收源文件片段和当前字段映射表
2. 构造抽取 prompt
3. 调用 LLM，要求输出严格 JSON
4. 解析 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 必须大于 0
• status 必须是四个枚举值之一

然后还有一个非常重要的去重逻辑。同一家供应商可能一周内发两次报价表，第二次是修订版，如果直接入库就会出现重复记录。我的做法是在目标表里建唯一索引，字段组合是 (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 errors

3.5 人工确认与入库落库

清洗校验通过后，我设计了一个 human_confirmation 节点。这个节点在 LangGraph 里用到一个特性叫 interrupt，简单说就是让图在某个节点暂停，等待外部输入再继续。实际跑的时候，这批待确认记录是攒下来，在系统里生成一个待办任务，人工在网页上扫一眼，有问题就改，没问题就批量点“确认”。

有人可能会说这不是增加工作量吗？操过系统的人都懂，宁可加一道轻量的人工确认，也不要事后花三倍时间洗数据。我的经验是，人工只需要关注 flagged 的记录，也就是校验有警告但不至于拒收的那些，正常情况下几十条记录几十秒就确认完了，不算负担。

确认后的记录进入 write_to_db，这一步就完全是常规数据库操作了。先检查唯一索引，不重复的 insert，重复的进 failed_records。最后 log_result 节点把处理结果记录到日志表，整个流程结束。

4. 完整代码骨架：一个最小可跑的 LangGraph 入库图

下面给一个可以实际跑起来的最小骨架。这个骨架我把数据库简化成了 SQLite，LLM 调用做了接口抽象，方便你替换成自己的模型。

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", encoding="utf-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(chunk=chunk) 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(checkpointer=MemorySaver())

这段代码如果你之前跟着系列文章跑过前面的例子，读起来应该很顺。核心就三件事：定义状态类型，写节点函数，把节点连成图。我觉得 LangGraph 最好的体验就是，你不需要发明一个新的编程范式，节点就是普通函数，图就是函数之间的路由关系。

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

这一部分我直接整理成速查表，全部来自我实际跑流程时踩过的坑，不是网上抄来的。

这里我特别展开说两个。

第一个是“温度和随机性”的问题。我在抽取节点里一开始把 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 项目，过程可见性和结果正确性一样重要。