tahasiddiquii/llm-extraction-pipeline

# llm-extraction-pipeline **将 LLM 作为生产级抽取基础设施**，这不仅仅是调用模型的演示，而是 围绕模型的系统：规则与 LLM 的分流、带有验证/修复的结构化输出、 版本化 prompt、包含 token/成本计算的每次调用追踪以及**严格的成本上限**、带有 **精确测量的精确率/召回率**的标注评估、**供应商模型漂移检测**、基于 embedding 的 **实体解析**，以及一个 **Airflow 3** ETL DAG。 [](https://github.com/tahasiddiquii/llm-extraction-pipeline/actions/workflows/ci.yml)     ``` pip install -e ".[dev]" extract demo # narrated end-to-end run: pipeline -> eval -> drift ``` 一切都在**离线且确定性**地运行，无需任何 API 密钥。设置一个 环境变量 (`EXTRACT_PROVIDER=anthropic`) 即可通过相同的接口切换到真实模型。 ## 运行结果（您可以自行复现） `extract eval` 会根据标注集对每个字段进行评分，并以此**拦截 CI**。这些数字 是由工具打印的，而非手写，运行它您将得到相同的结果： | | 精确率 | 召回率 | F1 | | --- | --- | --- | --- | | **总体** (20 份文档) | **1.000** | **0.975** | **0.988** | | 规则路由 (JSON-LD + 带标签的 HTML) | 1.000 | 1.000 | 1.000 | | LLM 路由 (杂乱的文本) | 1.000 | 0.956 | 0.977 | 整个集合的成本：**$0.0075** ($0.00037/文档)。**40% 的文档从未触及模型**， 它们被免费解析了。召回率刻意**低于 1.0**：少数真正困难的文本 案例（“数千人的员工队伍”，“在伦敦运营”）被留为 `null` 而不是 盲目猜测。诚实的召回率 + 高精确率才是重点，请参阅 [reports/eval_report_example.md](reports/eval_report_example.md)。 ## 为什么这契合数据抽取岗位 | 工作要求 | 在这里是如何构建的 | 试一试 | | --- | --- | --- | | **判断力：规则 vs LLM** | `triage.py` 将 JSON-LD 和带标签的 HTML 发送给确定性解析器；只有文本才会升级提交给模型。每次路由的评估*证明*了这种拆分是值得的。 | `extract eval` | | **结构化、可信的输出** | `instructor` + Pydantic schema；`validate.py` 进行强制转换（`"1,200"` 转为 `1200`），将不可能的情况置为空（年份 `1300`，负数的员工人数），拒绝没有名称的记录。 | `extract extract --id d1` | | **Prompt 即基础设施** | `prompts.py`，固定的、版本化的评分标准 (`v1`/`v2`)，记录在每次追踪中，通过配置进行回滚。 | n/a | | **基于真实数学的成本控制** | `router.py` 根据固定费率表对每次调用进行定价，随着支出接近上限，**将能力强的模型降级为廉价的模型**，并在预算耗尽之前停止（路由到审查）。 | `extract trace` | | **可观测性** | 每次抽取都会生成追踪记录（路由、模型、prompt 版本、token、成本、延迟、有效性、置信度）；可选的 **Langfuse v4** (OTel 原生) 镜像。 | `extract trace` | | **供应商模型漂移** | `evals.py` 通过一个静默衰退的模型重新对黄金集进行评分，并标记每个字段的 F1 下降。 | `extract drift` | | **实体解析** | `dedup.py`：规范化 -> 分块 -> 模糊匹配 + embedding 评分 -> union-find -> 合并。 | `extract dedup` | | **幂等 ETL + 编排** | `pipeline.py` 隔离不良记录（绝不会导致批处理失败）并按业务键进行 upsert；`dags/` 是 Airflow 3 Task SDK DAG。 | `extract run --show` | ## 每个命令展示了什么 ``` extract extract --id d10 # one document -> structured JSON + its trace extract run --show # ETL: extract -> validate -> resolve -> idempotent load extract eval # field-level precision/recall/F1 + the CI gate extract drift # detect a silently-regressed model (exits non-zero on drift) extract dedup # entity resolution over the sample set extract trace # the cost / token / route ledger ``` `extract drift` 输出，固定的模型别名发生衰退并被捕获： ``` stable F1 0.988 -> drifted F1 0.934 DRIFT DETECTED: the pinned model alias regressed: · employee_count: F1 0.97 -> 0.70 (-0.27) · founded_year: F1 1.00 -> 0.73 (-0.27) ``` ## 工作原理 一份文档，一条路径；唯一的分支是规则与 LLM 的决策，并且它会被记录下来。 ``` flowchart LR A[RawDocument] --> B{triage} B -->|JSON-LD| C[parse_jsonld] B -->|labelled HTML| D[parse_semi] B -->|prose| E{router · cost ceiling} E -->|budget ok| F[LLM · structured output] E -->|over ceiling| R[human review] C --> V[validate + repair] D --> V F --> V V -->|invalid| Q[quarantine] V -->|valid| N[normalize] --> K{confidence} K -->|low| R K -->|ok| G[entity resolution] --> L[(idempotent load)] ``` 包含组件表的完整说明：[docs/architecture.md](docs/architecture.md)。 可复用的抽取契约（schema + 评分标准 + 路由规则）：[SKILL.md](SKILL.md)。 ## 基于最新的 2026 技术栈构建 真实的、固定的、当前的版本（已于 2026 年 7 月验证），作为受保护的代码路径连接，因此离线 默认设置保持无密钥状态，而生产环境可通过一个环境变量进行切换。 | 领域 | 库 | 版本 | | --- | --- | --- | | 模型 / 结构化输出 | `openai` · `anthropic` · `instructor` | 2.44 · 0.115 · 1.15 | | 反爬虫 Web 采集 | `scrapling` · `httpx` · `parsel` · `curl-cffi` | 0.4.9 · 0.28 · 1.11 · 0.15 | | 可观测性 | `langfuse` (v4, OTel 原生) | 4.12 | | 实体解析 | `rapidfuzz` · `sentence-transformers` · `qdrant-client` · `splink` | 3.14 · 5.6 · 1.18 · 4.0 | | 编排 | `apache-airflow` (Task SDK + Assets) | 3.2 | | 数据仓库 | `dlt` · `duckdb` | 1.28 · 1.5 | | 核心 | `pydantic` · `pydantic-settings` | 2.x | ## 设计决策与诚实的局限性 - **刻意追求精确率而非召回率。** 评分标准规定*宁可使用 `null` 也不要盲目猜测*。这使得 精确率保持在 1.0，并留下一些难以处理的文本案例未解决。伪造这些结果会使评估 变成谎言。 - **Mock 是一个真实的解析器，而不是随机数据。** `providers.py` 确定性地将 “几十个”解析为 36，将“2000 年代初”解析为 2001，因此离线评估能够反映模型的行为，并且 漂移测试通过*移除*该能力来工作。 - **采集是一个接口，受道德约束。** 反爬虫扩展是用于 JS 渲染 / 代理 / 自愈选择器的前门；任何真实的爬取都必须遵守 `robots.txt`、网站 ToS 和速率限制。不发布任何第三方内容，样本数据均为合成且原创。 - **Airflow 是可选的。** DAG 是基于 Airflow 3 Task SDK 编写的，并在本地使用真实的 Airflow 进行了测试；CI 运行更轻量级的离线路径 (`importorskip`) 以保持快速。 ## 开发 ``` python -m venv .venv && source .venv/bin/activate pip install -e ".[dev]" # add ,airflow to run the DAG test locally make ci # ruff + pytest + eval gate (exactly what CI runs) ``` 78 个测试覆盖了分流、解析器、provider（包括漂移）、验证/修复、成本上限 router、抽取、evals + 拦截、dedup、幂等 pipeline、追踪、CLI 以及 Airflow DAG。Lint 使用 ruff；eval 拦截是 `make ci` 和 pre-push hook 的一部分。 ## 更广泛的作品集的一部分 此仓库构建所基于的平台基础组件和应用系统： - [ai-harness](https://github.com/tahasiddiquii/ai-harness) · [llm-eval-observability](https://github.com/tahasiddiquii/llm-eval-observability) · [llm-guardrails-redteam](https://github.com/tahasiddiquii/llm-guardrails-redteam) · [hybrid-graph-rag](https://github.com/tahasiddiquii/hybrid-graph-rag) - [llm-router](https://github.com/tahasiddiquii/llm-router) · [drift-detection](https://github.com/tahasiddiquii/drift-detection) · [support-copilot](https://github.com/tahasiddiquii/support-copilot) · [invoice-ap-agent](https://github.com/tahasiddiquii/invoice-ap-agent) · [deep-research-agent](https://github.com/tahasiddiquii/deep-research-agent) · [vendor-risk-agent](https://github.com/tahasiddiquii/vendor-risk-agent) - [timeseries-forecasting](https://github.com/tahasiddiquii/timeseries-forecasting) · [timeseries-classification](https://github.com/tahasiddiquii/timeseries-classification) · [tabular-ml](https://github.com/tahasiddiquii/tabular-ml) · [neural-net-from-scratch](https://github.com/tahasiddiquii/neural-net-from-scratch) ## 许可证 MIT © 2026 Taha Siddiqui

标签：Airflow, DLL 劫持, ETL流水线, LLM路由, 信息抽取, 大语言模型, 实体消歧, 数据提取, 运行时操纵, 逆向工具