我要投稿
微信搜一搜，关注“活水智能”
AI代理如何让文档知识图谱构建更高效？本文揭秘模块化处理流程的关键突破。 核心内容： 1. AI代理如何解决传统LLMs在知识图谱构建中的效率瓶颈 2. 结构化输出对文档层级解析的革命性优化 3. 实际案例展示单元-章节-实体模型的简化实现路径
本文将介绍如何通过引入AI代理（Agent）来简化并模块化这一流程。
使用代理的一个关键优势在于它能够生成结构化输出，这极大地简化并优化了层级文档的解析和处理过程。
我们先回顾一下本体（Ontology）模式：
文章中定义了四个模型：单元（Unit）、章节（Section）、子章节（Sub Section）以及实体（Entity）。但出于简化考虑，本文仅讨论其中的单元、章节和实体三种模型。
这是此前的方案：
我的数据是一系列 .txt 文件，每个文件都对应一本书中的一个单元，这些单元是通过光学字符识别（OCR）过程提取出来的。例如：
Political, Economic, and Cultural Situation under the Nguyễn Dynasty (First Half of the 19th Century)
Introduction
In 1802, the Nguyễn Dynasty was established. During its half-century of rule over a country that had just emerged from major upheavals, the Nguyễn court focused on consolidating power, restoring the economy, and reorganizing cultural affairs. However, entrenched in conservative feudal ideology, the Nguyễn regime failed to lay the foundation for a new phase of development.
Government Organization and Foreign Policy
After defeating the Tây Sơn Dynasty, Nguyễn Ánh proclaimed himself emperor in 1802, taking the reign title Gia Long and founding the Nguyễn Dynasty, with the capital in Phú Xuân (present-day Huế). In 1804, the country was officially renamed "Việt Nam," though it was later changed to "Đại Nam."
The central government followed the Lê Dynasty model but increased the emperor's authority. To manage the newly unified country, Gia Long initially divided the territory into three regions: Bắc Thành (Northern provinces), Gia Định Thành (Southern provinces), and Trực Doanh (Central provinces under direct royal administration). Each region had a Governor-General. In 1831–1832, Emperor Minh Mạng abolished these divisions and reorganized the country into 30 provinces and 1 special prefecture (Thừa Thiên), with officials such as Tổng đốc (Governor-General) and Tuần phủ (Provincial Governor) appointed by the royal court. Subdivisions like districts, counties, and villages remained unchanged.
Initially, officials were chosen from Nguyễn Ánh's former followers. Later, Confucian education and civil examinations became the main source of recruitment. Salaries were set, but officials were not granted land, and corruption gradually emerged among many bureaucrats.
A new legal code called Hoàng Việt Luật Lệ (also known as the Gia Long Code) was issued, containing nearly 400 articles that strictly protected the monarchy and feudal order.
...
Economic Situation and Nguyễn Policies
In the early 19th century, the country experienced temporary peace and unity. While the economy had favorable conditions, it also faced many difficulties.
Agriculture remained backward and unchanged. Much land was left fallow. In 1804, the state reinstated the "quân điền" (equal-field system), but public land only accounted for 20% of the total. Moreover, land distribution prioritized nobles, officials, and soldiers.
The court encouraged land reclamation through different means: allowing free settlement or providing funds and equipment (tools, buffalo) for farming. Despite efforts, land expansion was modest.
The state invested annually in irrigation, dredging canals, and reinforcing dykes, but these efforts were insufficient to prevent flooding.PydanticAI
我将使用 PydanticAI 框架来构建一个多代理系统。如果您不熟悉该框架，建议查阅其官方文档。
大型语言模型（LLMs）
本项目中，我将在 Google Colab 上部署 Ollama，并通过 Ngrok 在本地进行连接。感兴趣的读者可参考以下链接获取详细操作指南：https://medium.com/@jupyter267/learn-to-deploy-llm-apps-run-ollama-on-google-colab-and-connect-locally-with-ngrok-4b84a1106270。具体来说，我将使用 Qwen3 模型。
项目结构：
src/
│
├── agents/ # 负责处理流程中特定任务的AI代理
│ ├── __init__.py
│ ├── _agent_registry.py # 用于管理代理实例的注册器
│ ├── _base.py # 所有代理的基础LLM模型
│ ├── doc_parser_agent.py # 用于文档解析的代理
│ ├── dynamic_relation_extractor_agent.py # 用于动态提取关系的代理
│ ├── section_distiller_agent.py # 用于提炼文档章节的代理
│ └── single_distiller_agent.py # 单代理提炼系统
│
├── config/ # Ollama配置设置
│ ├── __init__.py
│
├── models/ # 数据模型、枚举和模式定义
│ ├── __init__.py
│ ├── enum.py # 项目中使用的自定义枚举
│ ├── schema.py # Pydantic或自定义模式定义
│ └── state.py # 状态管理模型
│
├── nodes/ # 代理或管道使用的模块化处理节点
│ ├── __init__.py
│ ├── doc_parser.py # 文档解析的节点逻辑
│ ├── section_distiller.py # 章节提炼的节点逻辑
│ └── single_distiller.py # 单代理提炼的节点逻辑
│
├── prompts/ # LLM代理使用的提示模板
│ ├── __init__.py
│ ├── doc_parser_prompt.py # 文档解析代理的提示
│ ├── dynamic_relation_extractor_prompt.py # 关系提取器的提示
│ ├── section_distiller_prompt.py # 章节提炼器的提示
│ └── single_distiller_agent.py # 单代理（初步方案）的提示
│
├── __init__.py
├── _utils.py # 实用函数
└── distill.py # 运行提炼流程的主脚本或入口点首先，我将尝试使用一种简单的一步（One-shot）解决方案来处理这个问题。如前所述，代理框架能够生成结构化输出，这使其非常适合此任务。我们的主要任务是设计一个与本体模式对齐的输出结构。
在 models/schema.py 文件中，我将定义模式中的模型：
class Mention(BaseModel):
type: Literal["LOCATION", "TIME", "PERSON", "EVENT", "ORGANIZATION"]
string: str = Field(...,
description="命名实体（如地点、时间、人物或事件）在原文中出现的精确字符串。")
classRelation(BaseModel):
head: str = Field(..., description="主实体（提及）。")
tail: str = Field(..., description="尾实体（提及）。")
relation_description: str = Field(...,
description="对主实体和尾实体之间关系的简要描述。")
classSection(BaseModel):
title: str = Field(..., description="章节标题")
summary: str = Field(...,
description="一段简要的总结（150-300字），突出本节涵盖的要点。")
mentions: List[Mention] = Field(...,
description="本节中发现的实体的完整列表。")
relations: List[Relation] = Field(...,
description="本节中提及实体之间关系的完整列表。")
@dataclass
classDistilledUnit:
title: str = Field(..., description="单元标题")
summary: str = Field(..., description="单元内容的简洁摘要，字数在150到300字之间。")
sections: List[Section] = Field(..., description="单元内容中包含的“章节”列表。")注意：
接着在 agent/single_distiller_agent.py 文件中：
from src.models import AgentName, DistilledUnit
from src.prompts import SINGLE_DISTILLER_AGENT
from pydantic_ai import Agent
from ._base import ollama_model
agent = Agent(
name=AgentName.single_distiller_agent.value,
model=ollama_model,
system_prompt=SINGLE_DISTILLER_AGENT,
result_type=DistilledUnit,
retries=3 # 若输出无法被解析为 DistilledUnit 类型，则最多尝试重试三次
)运行代理：
result = await agent.run(raw_document)
为什么不选择这种方法：
Relation 对象中的 head 和 tail 字段必须与 Mention 列表中找到的字符串完全匹配。这一严格要求至关重要，因为在知识图谱构建阶段，我需要将每条关系映射回其对应的提及，以准确识别所涉实体。在许多现有方法中，通常是先提取关系，然后再从 head 和 tail 中派生出提及。然而，仅仅依赖关系提取可能会导致遗漏文本中出现的许多实体提及。这在历史或医学等领域尤其成问题，因为全面提取提及可以显著提高信息检索等下游任务的性能。鉴于一步到位方案存在局限性，本文提出了一种多阶段处理流程，其中每个阶段都由一个专门的代理负责。在这种设计中，每个代理负责流程中的一个特定子任务（例如，提及提取、章节提炼、关系提取），这使得整个过程更具模块化、可控性和可解释性。
系统包含以下三个核心代理：
# ._utils.py
import anyio
from src.models import MyUsage
T = TypeVar("T")
asyncdeftask_group_gather(tasks: Sequence[Callable[[], Awaitable[T]]]):
"""异步运行任务"""
results: list[T] = [None] * len(tasks)
print('len result', len(results))
asyncdef_run_task(tsk: Callable[[], Awaitable[T]], index: int):
"""运行任务并将结果存储到正确索引的辅助函数。"""
results[index] = await tsk()
asyncwith anyio.create_task_group() as tg:
for i, task inenumerate(tasks):
tg.start_soon(_run_task, task, i)
return results
defupdate_usage(current_usage: MyUsage, new_usage: Usage):
current_usage.requests += new_usage.requests
current_usage.request_tokens += new_usage.request_tokens
current_usage.response_tokens += new_usage.response_tokens
current_usage.total_tokens += new_usage.total_tokens
return current_usage# models/enum.py
from enum import Enum
# 创建一个枚举对象来管理代理，这样我们就不必硬编码
classAgentName(Enum):
single_distiller_agent: str = "single_distiller_agent"
doc_parser_agent: str = "doc_parser_agent"
section_distiller_agent: str = "section_distiller_agent"
# models/schema.py
# 最终目标数据对象是 `DistilledUnit`
from pydantic import BaseModel, Field, create_model
from dataclasses import dataclass
from typing importLiteral, List, Optional, Any, Type
classMention(BaseModel):
type: Literal["LOCATION", "TIME", "PERSON", "EVENT", "ORGANIZATION"]
string: str = Field(...,
description="命名实体（如地点、时间、人物或事件）在原文中出现的精确字符串。")
classRelation(BaseModel):
head: str = Field(..., description="主实体（提及）。")
tail: str = Field(..., description="尾实体（提及）。")
relation_description: str = Field(...,
description="对主实体和尾实体之间关系的简要描述。")
classSection(BaseModel):
title: str = Field(..., description="章节标题")
summary: str = Field(...,
description="一段简要的总结（150-300字），突出本节涵盖的要点。")
mentions: List[Mention] = Field(...,
description="本节中发现的实体的完整列表。")
relations: Optional[List[Any]] = None
classSectionContent(BaseModel):
title: str = Field(..., description="章节标题")
content: str = Field(..., description="章节内容")
classUnit(BaseModel):
title: str = Field(..., description="单元标题")
summary: str = Field(..., description="单元内容的简洁摘要，字数在150到300字之间。")
sections: List[SectionContent] = Field(..., description="单元内容中包含的“章节内容”列表（按顺序）。")
@dataclass
classDistilledUnit:
title: str = Field(..., description="单元标题")
summary: str = Field(..., description="单元内容的简洁摘要，字数在150到300字之间。")
sections: List[Section] = Field(..., description="单元内容中包含的“章节”列表。")
# 该方法适用于后续场景，通过创建一个动态数据模型，其中主实体（head）和尾实体（tail）均取自提及列表的实际值，从而确保代理能够生成一致的提及字符串。
defbuild_dynamic_relation_model(mention_strings: List[str]) -> BaseModel:
mention_literals = Literal[tuple(mention_strings)]
DynamicRelation = create_model(
"DynamicRelation",
head=(mention_literals, Field(..., description="提及的实体（主实体）。")),
tail=(mention_literals, Field(..., description="提及的实体（尾实体）。")),
relation_description=(str, Field(..., description="对主实体和尾实体之间关系的简要描述。")),
)
return DynamicRelation
# models/state.py
from dataclasses import dataclass
@dataclass
classMyUsage:
requests: int | None = 0
request_tokens: int | None = 0
response_tokens: int | None = 0
total_tokens: int | None = 0
# 实际上，我第一次使用 PydanticAI 时，也曾疑惑何时使用 Dependencies/State。
# 我认为，对于需要在代理之间保存/共享的数据，您可能需要使用 State。
# 因为在运行流程结束后，您仍然可以访问 State 的值。
# 其他例如连接/数据库等，您可能需要使用 Dependencies。为了管理所有代理，我按照单例工厂模式创建了一个 AgentRegistry（代理注册器）：
# agents/_agent_registrt.py
from pydantic_ai import Agent
from typing importCallable
from src.models import AgentName
classAgentRegistry:
_agents: dict[AgentName, Agent] = {}
@classmethod
defregister(cls, agent_name: AgentName):
"""一个用于自动注册代理的装饰器"""
defdecorator(func: Callable) -> Callable:
cls._agents[agent_name] = func()
print("已注册代理", agent_name)
return func
return decorator
@classmethod
defget(cls, agent_name: AgentName) -> Agent:
return cls._agents[agent_name]# agents/doc_parser_agent.py
from ._agent_registry import AgentRegistry
from src.models import AgentName, Unit
from src.prompts import DOC_PARSER_PROMPT
from pydantic_ai import Agent
from ._base import ollama_model
@AgentRegistry.register(AgentName.doc_parser_agent)
defcreate_doc_parser_agent() -> Agent:
agent = Agent(
name=AgentName.doc_parser_agent.value,
model=ollama_model,
system_prompt=DOC_PARSER_PROMPT,
result_type=Unit,
retries=3
)
return agent# agents/section_distiller_agent.py
from ._agent_registry import AgentRegistry
from src.models import AgentName, Section
from src.prompts import SECTION_DISTILLER_AGENT
from pydantic_ai import Agent
from ._base import ollama_model
@AgentRegistry.register(AgentName.section_distiller_agent)
defcreate_section_distiller_agent() -> Agent:
agent = Agent(
name=AgentName.section_distiller_agent.value,
model=ollama_model,
system_prompt=SECTION_DISTILLER_AGENT,
result_type=Section,
retries=3
)
return agent# agents/dynamic_relation_extractor_agent
from pydantic import BaseModel
from src.prompts import DYNAMIC_RELATION_EXTRACTOR_PROMPT
from pydantic_ai import Agent
from typing importList
from ._base import ollama_model
defcreate_dynamic_relation_extractor_agent(
relation_model: BaseModel,
mention_strings: List[str]
)->Agent:
relation_extractor_agent = Agent(
name="relation_extractor_agent",
model=ollama_model,
retries=5,
output_type=List[relation_model],
system_prompt=DYNAMIC_RELATION_EXTRACTOR_PROMPT.format(mentions=mention_strings)
)
return relation_extractor_agent文档解析节点（Doc Parser Node）：
# nodes/doc_parser.py
from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End
from src.models import MyUsage, AgentName, Unit, DistilledUnit
from src.agents import AgentRegistry
from src.nodes.section_distiller import SectionDistillerNode
from src._utils import update_usage
from typing importAny
from src._utils import task_group_gather
@dataclass
classDocParserNode(BaseNode[MyUsage, None, Any]): # Generic[StateT, DepsT, NodeRunEndT])
raw_document: str
asyncdefrun(
self,
ctx: GraphRunContext[MyUsage, None],
) -> End:
result = await AgentRegistry.get(
AgentName.doc_parser_agent
).run(self.raw_document, model_settings={"temperature": 0})
# 每次调用代理时更新总使用量
usage = result.usage()
ctx.state = update_usage(ctx.state, usage)
unit: Unit = result.output
sections = unit.sections
# 异步处理多个章节
results = await task_group_gather(
[
lambda section=section: SectionDistillerNode(section=section).run(ctx)
for section in sections
]
)
# 最终结果
distilled_unit = DistilledUnit(
title=unit.title,
summary=unit.summary,
sections=[item.data for item in results],
)
return End(data=distilled_unit)章节提炼节点（Section Distiller Node）：
# nodes/section_distiller.py
from dataclasses import dataclass
from pydantic_graph import BaseNode, GraphRunContext, End
from src.models import MyUsage, AgentName, Section, SectionContent, Mention, build_dynamic_relation_model
from src.agents import AgentRegistry, create_dynamic_relation_extractor_agent
from src._utils import update_usage
from typing importAny
@dataclass
classSectionDistillerNode(BaseNode[MyUsage, None, Any]):
section: SectionContent
asyncdefrun(self, ctx: GraphRunContext[MyUsage, None]) -> End:
# 提取信息：摘要、提及、标题
distilled_result = await AgentRegistry.get(
AgentName.section_distiller_agent
).run(user_prompt=f"section_title:{self.section.title}\n section_content: {self.section.content}",
model_settings={"temperature": 0})
usage = distilled_result.usage()
ctx.state = update_usage(ctx.state, usage)
distilled_section: Section = distilled_result.output
mentions = distilled_section.mentions
mention_strings = [item.string for item in mentions]
# 调用关系提取代理提取关系
# 构建关系数据模型
relation_model = build_dynamic_relation_model(mention_strings=mention_strings)
# 构建关系提取代理
relation_extractor_agent = create_dynamic_relation_extractor_agent(
relation_model=relation_model,
mention_strings=mention_strings,
)
relation_result = await relation_extractor_agent.run(
user_prompt=f"section_title:{self.section.title}\n section_content: {self.section.content}",
model_settings={"temperature": 0}
)
usage = relation_result.usage()
ctx.state = update_usage(ctx.state, usage)
distilled_section.relations = relation_result.output
return End(distilled_section)至此，我们便可以创建 Graph 对象来运行工作流了。
import asyncio
import json
from pydantic_graph import Graph
from src.nodes import SectionDistillerNode, DocParserNode
from src.models import MyUsage
graph = Graph(nodes=[DocParserNode, SectionDistillerNode])
state=MyUsage()
asyncdefmain():
result = await graph.run(DocParserNode(sample), state=state)
print(result)召回率（Recall）
print("=========")
print(result.state)
asyncio.run(main())为了评估和比较不同框架或方法的性能，我们需要标注过的测试用例。对于这类信息提取任务，召回率（Recall） 是衡量系统检测或提取信息准确性的有效指标。然而，在本文中，我将只分析上述两种方法在一个单一示例上的输出。
state=MyUsage(requests=1, request_tokens=2803, response_tokens=3276, total_tokens=6079))
state=MyUsage(requests=11, request_tokens=17836, response_tokens=24076, total_tokens=41912))
在本文中，我介绍了如何通过用模块化、基于代理的设计取代繁重的LLM提示工程，来简化文档到图谱的构建流程。通过使用能够生成结构化输出的代理，我们使层级文档的解析和处理过程更加可靠和易于维护。这一转变，为构建更清晰、更具可扩展性的文档中心知识图谱奠定了基础。
至此，我们完成了对文档知识图谱构建流程的改进介绍。感谢您的阅读！
53AI，企业落地大模型首选服务商
产品：场景落地咨询+大模型应用平台+行业解决方案
承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业
2025-05-06
2025-05-23
2025-05-15
2025-05-28
2025-05-22
2025-05-07
2025-05-20
2025-05-09
2025-05-21
2025-05-20
回到顶部
粤ICP备14082021号