openJiuwen agent-core 三元组抽取器（TripleExtractor）实战指南：基于 LLM 的 OpenIE 抽取与校验

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

导读

TripleExtractor是 openJiuwen agent-core 检索（retrieval）模块中基于大语言模型（LLM）实现的开源信息抽取（OpenIE）三元组抽取器，负责把文本分块（TextChunk）转换成 RDF 风格的<subject, predicate, object>三元组，为 GraphRAG 图索引与图谱检索提供数据底座。本文以 TripleExtractor API 文档 为主体，结合 triple_extractor.py 源码、单元测试 与 GraphKnowledgeBase 调用链，完整讲解其初始化参数、抽取流程、可选校验机制、提示词模板、结果解析与错误处理，并给出可直接落地的接入示例。

1. 功能定位与设计背景

TripleExtractor属于索引流水线（indexing pipeline）中 Processor 体系的一环。整个抽取器体系继承结构如下：

• Processor：所有处理器（Parser、Chunker、Extractor）的抽象基类，定义process()方法，见 processor/base.py；
• Extractor：抽取器抽象基类，在process()中直接转发给extract()，见 extractor/base.py；
• TripleExtractor：本文主角，实现 OpenIE 三元组抽取及可选校验；
• OntologyTripleExtractor：同目录下的本体约束变体，支持按ontology_classes约束实体类别，见 ontology_triple_extractor.py。

从源码结构看，TripleExtractor是通用型抽取器：不依赖外部图数据库，仅通过 LLM 从文本中抽取实体与关系，因此可嵌入GraphKnowledgeBase（graph_knowledge_base.py）等图增强知识库，作为"块索引 + 三元组索引"双索引中的三元组生产者。

2. 类签名与初始化参数

根据 API 文档与 triple_extractor.py#L27-L50，构造签名如下：

TripleExtractor( llm_client: Any, # LLM 客户端实例，必填 model_name: str, # 模型名称，必填 temperature: float = 0.0, # 采样温度，默认 0.0 max_concurrent: int = 50, # 最大并发数，默认 50 validate: bool = False, # 是否启用 LLM 校验，默认 False **kwargs: Any # 预留扩展参数 )

各参数含义与底层影响：

源码中的初始化行为（triple_extractor.py#L46-L50）：

self.llm_client = llm_client self.model_name = model_name self.temperature = temperature self.limiter = asyncio.Semaphore(max_concurrent) self.validate = validate

3. 核心方法：extract

extract是唯一的公开异步入口，签名与行为见 triple_extractor.py#L52-L80：

async def extract( self, chunks: List[TextChunk], **kwargs: Any, ) -> List[Triple]

• 输入：List[TextChunk]，即文本分块列表，例如[TextChunk(...), TextChunk(...)]；
• 输出：List[Triple]，合并所有成功分块抽取出的三元组，例如[Triple(...), Triple(...)]；
• 行为：先执行_extract_internal()并行抽取；若构造时validate=True，再执行_validate_internal()逐块校验并过滤；
• 异常：任一分块失败时，按分块顺序抛出第一个错误；BaseError原样上抛，其他异常被包装为BaseError。

TextChunk与Triple的数据模型分别定义在 document.py 与 triple.py：

• TextChunk：包含id_（分块 ID）、text（文本内容）、doc_id（父文档 ID）、metadata（元数据字典）与可选embedding；
• Triple：Pydantic 模型，字段为subject、predicate、object三个字符串，外加metadata字典（抽取时写入doc_id与chunk_id，用于追溯三元组的来源）。

3.1 两阶段执行逻辑

extract()的完整执行链路（源码见 triple_extractor.py#L169-L272）：

1. 抽取阶段_extract_internal：为每个分块创建 asyncio 任务_extract_chunk，任务内先获取信号量，再调用_build_prompt()构造提示词并交给_invoke_and_parse()调用 LLM、解析结果；
2. 校验阶段_validate_internal：仅当validate=True且存在三元组时触发。先把候选三元组按metadata["chunk_id"]分组（defaultdict(list)），再对每个有候选三元组的分块发起校验 LLM 调用，校验提示词由_build_validation_prompt()构造；
3. 结果汇聚_gather_results：用asyncio.gather(..., return_exceptions=True)等待全部任务，将成功结果扁平化合并；记录第一个异常及其对应分块 ID，最后统一抛出（BaseError原样上抛，普通异常包装为RETRIEVAL_KB_TRIPLE_EXTRACTION_PROCESS_ERROR）。

4. LLM 调用与结果解析

4.1 调用方式

LLM 交互发生在_invoke_and_parse()（triple_extractor.py#L83-L117）：

messages = [{"role": "user", "content": prompt}] completion = await self.llm_client.invoke( messages=messages, temperature=self.temperature, ) triples, parse_success = self._parse_triples( completion.content, chunk.doc_id, chunk.id_ )

即：构造单条 user 消息 → 异步调用invoke→ 用_parse_triples解析模型返回文本。解析失败时抛出RETRIEVAL_KB_TRIPLE_EXTRACTION_PROCESS_ERROR，错误消息形如"{chunk.id_}: LLM response could not be parsed as valid triple JSON"。

4.2 解析规则（_parse_triples）

解析器位于 triple_extractor.py#L443-L516，对模型输出非常宽容：

• 先strip()去除首尾空白；若以 ``` 开头，则剥离首尾围栏行（兼容 markdown 围栏 JSON）；
• 使用json_repair.repair_json(content, return_objects=True)修复模型常见的破损 JSON；
• 兼容两种顶层结构：带triples键的字典，或直接是三元组数组的列表；
• 逐条校验：跳过非列表/元组项、长度不足 3 的项、以及前三个元素包含嵌套结构或None的项；三元组只取前三个元素，忽略多余字段；
• 每个合法三元组构造Triple(subject=..., predicate=..., object=..., metadata={"doc_id": doc_id, "chunk_id": chunk_id})，三个字段均strip()；
• 返回(triples, parse_success)：空三元组列表被视为解析成功（([], True)），而硬解析失败返回([], False)；若存在非法项则以 warning 日志记录被忽略的数量。

以下测试用例（test_triple_extractor.py#L147-L199）精确印证了解析行为：

• '[["a", "b", "c"]]'→ 成功，得到 1 条三元组("a", "b", "c")；
• '[["a", "b", "c", "ignored", 99]]'→ 成功，多余字段被忽略，object仍为"c"；
• '{"triples": [["x", "y", "z"]]}'→ 字典包裹形式解析成功；
• '{"named_entities": ["Alice", "Bob"], "triples": [["Alice", "knows", "Bob"]]}'→ 只取triples，命名实体列表不参与三元组构造；
• '{"named_entities": ["Alice", "Bob"]}'→ 缺少triples键，解析失败；
• '{"triples": [["a", "b", "c"], ["x"], {"bad": 1}, ["y", ["nested"], "z"]]}'→ 非法项被忽略，仅保留合法 1 条；
• '{"triples": [["x"], {"bad": 1}]}'→ 全部非法，解析失败。

5. 提示词模板：抽取与校验两套 Prompt

提示词构造是纯代码内嵌模板，不依赖外部配置文件，源码见 triple_extractor.py#L274-L441。

5.1 抽取提示词_build_prompt

输入为passage（分块正文）与可选title（取自chunk.metadata.get("title", "")），标题为空时回退为"Untitled"。模板要点：

• 任务定义：要求模型根据给定标题与段落构建 RDF 风格图，抽取命名实体与关系，输出恰好一个合法 JSON 对象；
• 输出格式：顶层必须且只能包含两个键"named_entities"（字符串数组）与"triples"（数组），其中每条三元组必须是恰好三个字符串组成的 JSON 数组；
• 质量要求：尽量解析代词为具体名字；优先使用至少包含一个（最好两个）来自原文的命名实体的三元组；实体与谓词措辞保持与源语言一致；不输出重复三元组；无三元组时返回{"named_entities": [...], "triples": []}；
• 两个 Few-shot 示例：分别以 NBA 球星 Magic Johnson 与游戏 Elden Ring 为演示，展示实体命名与关系谓词（如"drafted by"、"director"）的规范化写法。

模板占位符最终通过prompt_template.format(passage=passage, title=title or "Untitled")填充。

5.2 校验提示词_build_validation_prompt

当validate=True时使用。输入为原文段落与候选三元组，候选三元组序列化方式：

triples_text = json.dumps( [[t.subject, t.predicate, t.object] for t in triples], ensure_ascii=False, indent=2, )

校验指令要求模型扮演 OpenIE 三元组校验器：

• 仅输出被原文直接支持的三元组；
• 允许为了正确性或清晰度修改谓词；
• 丢弃依赖外部知识、日期/数字/地点不匹配、或无法由文本必然成立的三元组；
• 输出格式为只含"triples"键的单个 JSON 对象，无合法三元组时返回{"triples": []}。

5.3 校验的调用代价

测试用例 test_extract_multiple_chunks_with_validation 验证：2 个分块在validate=True时，llm_client.invoke被调用4 次（每块抽取 1 次 + 校验 1 次），即校验模式会把 LLM 调用量翻倍。因此在吞吐敏感场景下，validate默认关闭是合理选择；追求图质量时再显式开启。

6. 错误处理与并发控制

6.1 并发控制

• 构造时创建asyncio.Semaphore(max_concurrent)；
• 抽取与校验的每个分块任务都在async with self.limiter:内执行，保证任意时刻在途 LLM 请求数不超过max_concurrent；
• 所有分块任务通过asyncio.create_task并发调度，再由asyncio.gather(return_exceptions=True)汇聚，单个分块失败不会拖垮整批任务。

6.2 错误语义

统一错误码定义在 codes.py#L585-L588：

RETRIEVAL_KB_TRIPLE_EXTRACTION_PROCESS_ERROR = ( 155507, "retrieval kb_triple_extraction process error, reason: {error_msg}", )

三种典型失败路径：

测试用例 test_extract_with_exception 印证：当invoke抛出Exception("429 too many requests")时，extract会抛出 code 为155507的BaseError，且消息包含原始异常文本；test_extract_invalid_json 印证非法 JSON 同样映射到该错误码。

7. 在知识库流水线中的接入位置

7.1 与 GraphKnowledgeBase 的协作

GraphKnowledgeBase的构造函数接受extractor: Optional[Extractor]参数（graph_knowledge_base.py#L36-L63）。在其add_documents()中（graph_knowledge_base.py#L127-L153）：

• 当self.config.use_graph为真且提供了extractor时，先对分块调用await self.extractor.extract(chunks)；
• 若产出三元组，则创建名为kb_{kb_id}_triples的三元组索引，并把每条三元组转换为文本格式f"{subject} {predicate} {object}"的TextChunk写入索引，同时在metadata中保留原始三元组 JSON。

相关开关定义于 common/config.py：KnowledgeBaseConfig.use_graph（默认False，决定是否启用图索引），以及RetrievalConfig.use_graph/graph_expansion（控制检索阶段是否走图检索与图谱扩展）。

7.2 最小接入示例

结合源码接口，一个可运行的接入骨架如下（LLM 客户端以具备async invoke的对象为例）：

import asyncio from openjiuwen.core.retrieval.common.document import TextChunk from openjiuwen.core.retrieval.indexing.processor.extractor.triple_extractor import TripleExtractor async def main(): # 1. 构造抽取器：temperature 用 0 保证稳定，并发控制在 10 extractor = TripleExtractor( llm_client=llm_client, # 需要支持 async invoke(messages=..., temperature=...) model_name="your-model-name", temperature=0.0, max_concurrent=10, validate=True, # 开启二次校验，过滤不被原文支持的三元组 ) # 2. 准备分块 chunks = [ TextChunk(id_="1", text="Alice knows Bob and works at Company.", doc_id="doc_1"), TextChunk(id_="2", text="Charlie is a manager at Startup.", doc_id="doc_1"), ] # 3. 并行抽取（+ 校验），返回 List[Triple] triples = await extractor.extract(chunks) for t in triples: print(t.subject, t.predicate, t.object, t.metadata) # metadata 含 doc_id 与 chunk_id asyncio.run(main())

7.3 生产使用建议

• 稳定性优先：抽取场景默认temperature=0.0，避免同一文本反复抽取产生不一致的实体与谓词；
• 按吞吐调节并发：max_concurrent应根据模型服务限流阈值设置，默认 50 适合并发能力较强的服务；限流时（如 429）错误会被包装为BaseError抛出，建议在上游配合重试策略；
• 图质量 vs 成本权衡：validate=True会翻倍 LLM 调用，但能过滤幻觉式三元组（依赖外部知识、日期/数字/地点不符等），适合对图谱准确性要求高的场景；
• 容错解析：模型输出即使包裹在 markdown 围栏或存在轻微 JSON 破损，_parse_triples也会借助json_repair尽力修复并忽略非法条目，实测中只需保证顶层结构合法即可成功。

8. 总结

TripleExtractor以"一次抽取 + 可选一次校验"的两阶段 LLM 流水线，把非结构化的文本分块转化为带doc_id/chunk_id溯源信息的三元组列表，是 openJiuwen agent-core 图增强知识库（GraphRAG）索引的关键生产者。它通过信号量限流、并发任务汇聚、宽容 JSON 解析与统一错误码，兼顾了吞吐、容错与结果的可用性；validate开关则为需要在图质量与调用成本之间做出权衡的场景提供了明确抓手。读者可依据 triple_extractor.py、单元测试 与 GraphKnowledgeBase 深入验证以上全部行为。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习