各位同仁，各位对文档智能处理充满热情的开发者们：
欢迎大家来到今天的技术讲座。今天，我们将深入探讨一个前沿且极具挑战性的领域——Agentic Document Parsing。顾名思义，我们讨论的不再是简单的文本提取或基于规则的模式匹配，而是如何构建一个拥有自主决策能力的“智能代理”，让它像人类专家一样，逐页审视复杂的PDF文档，并能够根据上下文和内容类型，智能地决定何时需要调用强大的视觉模型进行深度解析。
一、 传统文档处理的困境与智能代理的崛起
长期以来，我们与文档打交道的方式，尤其是非结构化或半结构化文档，一直面临着诸多挑战。从简单的发票、合同到复杂的科研报告、财报，它们往往融合了文本、表格、图表、图像等多种信息载体。
传统的文档处理方案，无论是基于光学字符识别（OCR）进行文本提取，还是利用正则表达式、模板匹配进行结构化信息抽取，都存在着固有的局限性：
- 对布局变化的脆弱性：微小的布局调整可能导致整个解析流程失效。
- 对非文本信息的无力：图表、流程图、组织结构图等视觉元素蕴含着丰富的语义，但OCR只能将其视为像素，无法理解其内在关系和数据。
- 缺乏上下文理解：孤立地提取信息，难以把握文档的整体逻辑和篇章结构。
- 难以处理模糊和不确定性：面对低质量扫描件或非标准格式，传统方法往往束手无策。
随着大型语言模型（LLMs）和多模态模型（如GPT-4V、LLaVA）的飞速发展，我们迎来了解决这些困境的全新范式——智能代理（Agent）。一个智能代理不仅仅是一个模型调用器，它是一个具备感知、规划、行动和反思能力的实体。在文档解析的语境下，这意味着它能够：
- 感知（Perception）：读取PDF的每一页，获取原始文本、布局信息、图像区域等。
- 规划（Planning）：根据当前页面的内容和整体目标，决定下一步行动。
- 行动（Action）：调用相应的工具，例如文本提取工具、视觉模型分析工具。
- 反思（Reflection）：评估工具的输出，更新对文档的理解，并调整后续策略。
而“Agentic Document Parsing”的核心，正是利用这种代理范式，赋予系统自主决策的能力，使其能够判断何时需要将一个视觉元素（如图表）送入专业的视觉模型进行解析，从而摆脱人工干预和预设规则的束缚，实现更深层次的文档理解。
二、 Agentic Document Parsing 的核心架构与组件
要构建一个能够逐页审视PDF并自主决策的智能代理，我们需要一套协同工作的组件和清晰的架构。
核心思想： 代理通过循环观察、思考、行动来逐步理解文档。当代理在某一页“看到”潜在的视觉信息时，它会基于文本上下文和自身知识进行判断，如果认为该视觉信息对文档理解至关重要且无法通过纯文本方式获取，便会调用专门的视觉解析工具。
让我们来看一下这种系统的主要组成部分：
| 组件名称 | 职责 | 关键技术/工具 |
|---|---|---|
| PDF加载与预处理器 | 读取PDF文件，提取原始文本、页面布局、图像区域及它们的边界框。 | pypdf, pdfplumber, fitz (PyMuPDF) |
| 代理核心 (Orchestrator) | 整个系统的决策中心，负责管理状态、规划行动、调用工具、处理反馈。 | LLM (例如 GPT-4, Llama), 内存管理模块, 代理框架 (LangChain Agents, LlamaIndex Agents) |
| 文本处理工具 | 执行基于文本的操作，如：提取区域文本、关键词识别、文本摘要。 | 内置文本解析器, LLM for NLP任务 |
| 视觉解析工具 | 处理图像信息，如：识别图表类型、提取图表数据、描述图像内容。 | 多模态LLM (GPT-4V, LLaVA), 专业的图像分析模型 (例如用于表格识别、图表识别的CV模型) |
| 知识库与记忆模块 | 存储代理在解析过程中积累的知识和上下文信息，支持RAG。 | 向量数据库 (Chroma, Pinecone), 关系数据库, 简单的字典/列表 |
| 输出与结构化模块 | 将解析结果转化为结构化的数据格式（JSON, YAML, 知识图谱）。 | 自定义数据模型, 图数据库 (Neo4j) |
工作流程概览：
- 加载文档：PDF预处理器加载PDF，将其拆分为独立的页面，并对每页进行初步的文本和图像区域识别。
- 逐页迭代：代理核心按顺序处理每一页。
- 页面观察：对于当前页，代理获取其文本内容、检测到的图像区域及其元数据。
- 决策（LLM推理）：代理（通过LLM）分析当前页的观察结果，结合之前的上下文，判断：
- 该页是否包含重要的视觉信息？
- 纯文本信息是否足以理解该页？
- 是否需要调用视觉解析工具？如果是，调用哪个工具，以及带上什么指令？
- 执行行动：根据决策，代理调用相应的工具。
- 如果决定不需要视觉解析，则仅使用文本处理工具，提取关键信息。
- 如果决定需要视觉解析，则将图像区域裁剪出来，并附带针对性的指令（Prompt）发送给视觉解析工具。
- 结果整合与反思：代理接收工具的输出，将其整合到当前的理解中，更新内存，并根据结果调整后续的解析策略。
- 循环往复：重复步骤3-6，直到所有页面处理完毕。
- 最终输出：将所有页面的解析结果汇集，生成最终的结构化文档理解报告。
三、 深度实现：Agentic Document Parsing 的技术细节与代码实践
现在，让我们深入到具体的实现细节，通过代码片段来理解各个组件如何协同工作。
我们将使用Python作为开发语言，并结合一些流行的库。
A. PDF预处理与初始信息提取
首先，我们需要一个能够可靠地从PDF中提取文本和识别图像区域的工具。pdfplumber是一个非常强大的库，它能够以编程方式访问PDF的各种内部结构，包括文本、线条、矩形和图像。
import pdfplumber
from typing import List, Dict, Any
class PDFProcessor:
def __init__(self, pdf_path: str):
self.pdf_path = pdf_path
self.doc = None
def load_pdf(self):
"""加载PDF文档。"""
try:
self.doc = pdfplumber.open(self.pdf_path)
print(f"成功加载PDF: {self.pdf_path}")
except Exception as e:
print(f"加载PDF失败: {e}")
self.doc = None
def get_page_info(self, page_number: int) -> Dict[str, Any]:
"""
获取指定页面的文本内容、图像区域及其边界框。
返回一个字典，包含页面文本和图像列表。
"""
if not self.doc:
print("PDF文档未加载。")
return {}
if page_number < 0 or page_number >= len(self.doc.pages):
print(f"页码 {page_number} 超出范围。")
return {}
page = self.doc.pages[page_number]
page_text = page.extract_text()
images = []
# pdfplumber的images属性会返回页面上的所有图像对象
# 我们可以根据需要过滤或提取更多信息
for img in page.images:
# img字典包含 'x0', 'y0', 'x1', 'y1' (边界框) 以及其他元数据
# 例如 'width', 'height', 'stream', 'uri' 等
images.append({
"bbox": (img['x0'], img['y0'], img['x1'], img['y1']),
"object_type": img.get('object_type', 'image'), # 可能是image, figure等
"width": img.get('width'),
"height": img.get('height'),
"stream": img.get('stream') # 图像的原始流数据，后续可用于保存或传输
})
# 简单地识别潜在的表格（通常在pdfplumber中也可能被识别为rects或lines的组合）
# 这里我们更关注图像，但可以扩展
tables = []
for table in page.find_tables():
tables.append({
"bbox": (table.bbox[0], table.bbox[1], table.bbox[2], table.bbox[3]),
"text_content": table.extract_text()
})
return {
"page_number": page_number,
"text": page_text,
"images": images,
"tables": tables # 包含通过文本识别的表格信息，与视觉模型解析的表格区分
}
def close_pdf(self):
"""关闭PDF文档。"""
if self.doc:
self.doc.close()
print("PDF文档已关闭。")
# 示例用法
# if __name__ == "__main__":
# processor = PDFProcessor("example.pdf") # 假设你有一个example.pdf文件
# processor.load_pdf()
# if processor.doc:
# for i in range(len(processor.doc.pages)):
# page_info = processor.get_page_info(i)
# print(f"--- Page {page_info['page_number'] + 1} ---")
# print(f"Text length: {len(page_info['text'])}")
# print(f"Found {len(page_info['images'])} images.")
# for img in page_info['images']:
# print(f" Image bbox: {img['bbox']}")
# processor.close_pdf()
这段代码为我们提供了一个基础，能够逐页获取文本和图像区域。图像的stream属性非常关键，它允许我们获取原始图像数据，以便发送给视觉模型。
B. 代理的决策循环与工具定义
现在是核心部分：构建代理。代理将由一个LLM驱动，并配备一系列工具。
我们将定义一个Agent类，它将包含：
- 一个LLM实例，用于决策。
- 一个工具字典，存储所有可用的工具。
- 一个记忆模块，用于存储上下文。
工具的抽象：
每个工具都应该有一个清晰的描述，告诉LLM它能做什么，以及它需要哪些参数。
import base64
from io import BytesIO
from PIL import Image
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
# 模拟一个LLM接口
class MockLLM:
def __init__(self, model_name: str = "gpt-4-turbo"):
self.model_name = model_name
def chat(self, messages: List[Dict[str, str]], tools: Optional[List[Dict]] = None, tool_choice: Optional[str] = None) -> Dict[str, Any]:
"""
模拟LLM的聊天和工具调用功能。
在实际应用中，这里会调用OpenAI或其他服务商的API。
"""
# 简化模拟：根据prompt内容进行简单判断
last_message_content = messages[-1]['content']
# 模拟决策调用视觉模型
if "是否需要对图表进行详细分析" in last_message_content and "图表" in last_message_content:
# 简单模拟，如果文本中提到“图表”，就假装LLM决定调用视觉模型
print("[MockLLM] 模拟LLM决定调用 'visual_analysis_tool'...")
# 假设LLM决定调用 visual_analysis_tool
# 真实情况下，LLM会返回一个tool_calls对象
return {
"choices": [{
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"tool_calls": [{
"id": "call_mock_visual_tool_1",
"type": "function",
"function": {
"name": "visual_analysis_tool",
"arguments": '{"image_data_base64": "placeholder_base64", "analysis_prompt": "详细分析图表内容，并提取关键数据点和趋势。"}'
}
}]
}
}]
}
elif "提取本页的关键信息" in last_message_content:
print("[MockLLM] 模拟LLM决定调用 'extract_text_info_tool'...")
return {
"choices": [{
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"tool_calls": [{
"id": "call_mock_text_tool_1",
"type": "function",
"function": {
"name": "extract_text_info_tool",
"arguments": '{"text_content": "placeholder_text", "analysis_prompt": "提取本页的标题、摘要和任何重要结论。"}'
}
}]
}
}]
}
else:
print(f"[MockLLM] 模拟LLM进行文本回复: {last_message_content[:50]}...")
return {
"choices": [{
"finish_reason": "stop",
"message": {
"role": "assistant",
"content": "根据您的指示，我将继续处理文档。"
}
}]
}
# 定义一个抽象工具基类
class Tool(ABC):
def __init__(self, name: str, description: str, parameters: Dict):
self.name = name
self.description = description
self.parameters = parameters # JSON Schema 格式
@abstractmethod
def run(self, **kwargs) -> Any:
pass
def get_definition(self) -> Dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
# 文本信息提取工具
class TextInfoExtractionTool(Tool):
def __init__(self, llm: MockLLM):
super().__init__(
name="extract_text_info_tool",
description="从给定文本内容中提取关键信息，如标题、摘要、结论等。",
parameters={
"type": "object",
"properties": {
"text_content": {"type": "string", "description": "要分析的文本内容。"},
"analysis_prompt": {"type": "string", "description": "具体分析指令。"}
},
"required": ["text_content", "analysis_prompt"]
}
)
self.llm = llm
def run(self, text_content: str, analysis_prompt: str) -> Dict[str, Any]:
# 实际中会调用LLM进行文本分析
print(f"[TextInfoExtractionTool] 正在处理文本，指令: {analysis_prompt[:30]}...")
# 模拟LLM处理结果
response = self.llm.chat(
messages=[
{"role": "system", "content": "你是一个专业的文本分析助手。"},
{"role": "user", "content": f"请根据以下文本内容：n```n{text_content[:200]}...n```n{analysis_prompt}"}
]
)
# 简化返回
return {"extracted_info": f"模拟提取结果：{analysis_prompt} 从文本中。"}
# 视觉分析工具
class VisualAnalysisTool(Tool):
def __init__(self, vision_llm: MockLLM): # 实际中可能是一个专门的多模态LLM或CV模型
super().__init__(
name="visual_analysis_tool",
description="对图像（特别是图表、流程图等）进行详细分析，提取数据、描述关系或总结内容。",
parameters={
"type": "object",
"properties": {
"image_data_base64": {"type": "string", "description": "图像的base64编码字符串。"},
"analysis_prompt": {"type": "string", "description": "针对图像的具体分析指令，例如'提取图表中的数值'或'描述流程图的步骤'。"}
},
"required": ["image_data_base64", "analysis_prompt"]
}
)
self.vision_llm = vision_llm
def run(self, image_data_base64: str, analysis_prompt: str) -> Dict[str, Any]:
# 在这里，image_data_base64会被解码，然后发送给视觉模型
print(f"[VisualAnalysisTool] 正在分析图像，指令: {analysis_prompt[:30]}...")
# 模拟视觉模型调用，实际会调用如GPT-4V API
# 由于我们这里是MockLLM，它不会真的处理图片，只是模拟其行为
# 真实场景下，vision_llm.chat会包含图片数据
response = self.vision_llm.chat(
messages=[
{"role": "system", "content": "你是一个专业的图像分析助手，善于理解图表和流程图。"},
{"role": "user", "content": f"请根据以下图像（base64编码）进行分析：{image_data_base64[:50]}... n{analysis_prompt}"}
]
)
# 简化返回
return {"visual_analysis_result": f"模拟视觉分析结果：{analysis_prompt} 从图像中获取。"}
# Agent核心
class AgenticParser:
def __init__(self, pdf_processor: PDFProcessor, llm: MockLLM, vision_llm: MockLLM):
self.pdf_processor = pdf_processor
self.llm = llm
self.vision_llm = vision_llm # 可以是同一个LLM实例，如果它支持多模态
self.tools: Dict[str, Tool] = {
"extract_text_info_tool": TextInfoExtractionTool(self.llm),
"visual_analysis_tool": VisualAnalysisTool(self.vision_llm)
}
self.memory: List[Dict[str, Any]] = [] # 存储已处理页面的关键信息
self.full_document_output: List[Dict[str, Any]] = []
self.tool_definitions = [tool.get_definition() for tool in self.tools.values()]
def _get_llm_system_prompt(self) -> str:
"""定义LLM的系统角色和指令。"""
return (
"你是一个智能文档解析代理，任务是逐页审视PDF文档。 "
"对于每一页，你需要先提取文本内容和识别潜在的图像区域。 "
"然后，根据页面内容和文档的整体上下文，决定是否需要调用视觉分析工具来深入理解图表、流程图等。 "
"如果页面主要是文本，则使用文本提取工具。如果发现图表或关键视觉信息，且文本不足以理解，则调用视觉分析工具。 "
"你的目标是全面理解文档内容，并以结构化形式输出。 "
"在处理完一页后，将关键发现添加到你的记忆中，以便后续页面参考。"
)
def _prepare_page_observation(self, page_info: Dict[str, Any]) -> str:
"""
根据页面信息生成LLM的观察输入。
这里需要包含图像区域的存在信息，但不是图像本身。
"""
observation_str = f"当前处理页码: {page_info['page_number'] + 1}n"
observation_str += "--- 页面文本内容 (截断前500字) ---n"
observation_str += page_info['text'][:500] + ("..." if len(page_info['text']) > 500 else "") + "n"
if page_info['images']:
observation_str += f"页面检测到 {len(page_info['images'])} 个图像/图表区域。n"
for i, img in enumerate(page_info['images']):
observation_str += f" - 图像 {i+1}: 边界框 {img['bbox']}, 类型: {img['object_type']}n"
# 这里可以添加一些关于图像周围文本的上下文，帮助LLM决策
if page_info['tables']:
observation_str += f"页面检测到 {len(page_info['tables'])} 个潜在表格区域（通过文本布局识别）。n"
for i, tbl in enumerate(page_info['tables']):
observation_str += f" - 表格 {i+1}: 边界框 {tbl['bbox']}n"
# 加上记忆中的历史信息作为上下文
if self.memory:
observation_str += "n--- 历史解析发现 (最近3页) ---n"
for item in self.memory[-3:]: # 只看最近3页
observation_str += f" - 页 {item['page_number'] + 1}: {item['summary_of_findings']}n"
return observation_str
def _execute_tool_call(self, tool_call: Dict[str, Any], page_info: Dict[str, Any]) -> Any:
"""执行LLM指定的工具调用。"""
tool_name = tool_call['function']['name']
tool_args = json.loads(tool_call['function']['arguments'])
tool = self.tools.get(tool_name)
if not tool:
return f"错误: 代理尝试调用未知工具: {tool_name}"
# 特殊处理视觉工具，需要传入图像数据
if tool_name == "visual_analysis_tool":
# 假设LLM指定了要分析第1个图像（这里简化处理，实际需要LLM指定图像的索引或bbox）
# 或者可以要求LLM提供一个通用的prompt，然后代理遍历所有图像
if page_info['images']:
# 获取第一张图的原始数据，并base64编码
# 真实场景中，需要根据tool_args中的信息（如果LLM提供）来确定是哪张图
first_image_stream = page_info['images'][0]['stream']
image_bytes = BytesIO(first_image_stream.get_data())
# 确保图像是RGB模式，有些PDF图像可能是单色或调色板模式
try:
image_pil = Image.open(image_bytes).convert("RGB")
except Exception as e:
print(f"无法打开或转换图像: {e}. 尝试直接使用原始流数据。")
# 如果PIL无法处理，可能需要更底层的处理或告知LLM
return f"错误: 无法处理图像流 {e}"
buffered = BytesIO()
image_pil.save(buffered, format="PNG") # 或JPEG
image_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8')
tool_args['image_data_base64'] = image_base64 # 更新参数
else:
return "错误: 代理尝试调用视觉工具，但页面上没有图像。"
elif tool_name == "extract_text_info_tool":
tool_args['text_content'] = page_info['text'] # 文本工具直接使用页面文本
try:
return tool.run(**tool_args)
except Exception as e:
return f"工具 '{tool_name}' 执行失败: {e}"
def parse_document(self):
"""主解析循环，逐页处理文档。"""
if not self.pdf_processor.doc:
self.pdf_processor.load_pdf()
if not self.pdf_processor.doc:
print("无法加载PDF，退出解析。")
return
total_pages = len(self.pdf_processor.doc.pages)
print(f"开始解析文档，共 {total_pages} 页。")
for i in range(total_pages):
print(f"n--- 代理开始处理第 {i+1} 页 ---")
page_info = self.pdf_processor.get_page_info(i)
# 1. 代理观察当前页面
observation = self._prepare_page_observation(page_info)
messages = [
{"role": "system", "content": self._get_llm_system_prompt()},
{"role": "user", "content": f"请审视以下页面内容，并决定下一步行动：n{observation}"}
]
tool_outputs = []
final_page_findings = {"page_number": i, "text_summary": "", "visual_insights": []}
# 2. 代理决策（LLM推理）
try:
response = self.llm.chat(messages, tools=self.tool_definitions, tool_choice="auto")
# 模拟OpenAI API的tool_calls解析
tool_calls = response['choices'][0]['message'].get('tool_calls')
if tool_calls:
print(f"代理决定调用 {len(tool_calls)} 个工具。")
for tool_call in tool_calls:
tool_output = self._execute_tool_call(tool_call, page_info)
tool_outputs.append({
"tool_name": tool_call['function']['name'],
"output": tool_output
})
# 根据工具输出更新页面发现
if tool_call['function']['name'] == "extract_text_info_tool":
final_page_findings["text_summary"] = tool_output.get("extracted_info", "无文本摘要")
elif tool_call['function']['name'] == "visual_analysis_tool":
final_page_findings["visual_insights"].append(tool_output.get("visual_analysis_result", "无视觉分析结果"))
# 将工具输出反馈给LLM，进行下一步决策或总结 (可选，可以循环多次)
messages.append(response['choices'][0]['message']) # 将LLM的tool_calls加入对话历史
messages.append({
"role": "tool",
"tool_call_id": tool_call['id'],
"content": json.dumps(tool_output) # 工具输出通常是JSON
})
# 再次调用LLM进行总结或下一步行动
summary_response = self.llm.chat(messages, tools=self.tool_definitions, tool_choice="auto")
if summary_response['choices'][0]['message'].get('content'):
print(f"代理对工具输出的总结/后续: {summary_response['choices'][0]['message']['content']}")
# 这里可以更新 final_page_findings
final_page_findings["agent_reflection"] = summary_response['choices'][0]['message']['content']
else: # LLM没有选择工具，直接返回文本回复
agent_reply = response['choices'][0]['message'].get('content', '无回复内容')
print(f"代理直接回复: {agent_reply}")
final_page_findings["text_summary"] = agent_reply
except Exception as e:
print(f"代理决策或执行工具时发生错误: {e}")
final_page_findings["error"] = str(e)
# 3. 反思与记忆更新
self.memory.append({
"page_number": i,
"summary_of_findings": final_page_findings["text_summary"] + " " + " ".join(final_page_findings["visual_insights"])
})
self.full_document_output.append(final_page_findings)
print(f"--- 第 {i+1} 页处理完成 ---")
print(f"页面发现: {final_page_findings}")
self.pdf_processor.close_pdf()
print("n文档解析完成。")
return self.full_document_output
# 辅助函数：将图像保存到内存并base64编码
def get_image_base64_from_stream(image_stream_data) -> str:
"""
将pdfplumber提取的图像流数据转换为base64编码的PNG字符串。
"""
if not image_stream_data:
return ""
image_bytes = BytesIO(image_stream_data.get_data())
try:
image_pil = Image.open(image_bytes).convert("RGB")
buffered = BytesIO()
image_pil.save(buffered, format="PNG")
return base64.b64encode(buffered.getvalue()).decode('utf-8')
except Exception as e:
print(f"处理图像流失败: {e}")
return ""
import json
if __name__ == "__main__":
# 创建一个虚拟的PDF文件用于测试
# 实际应用中需要一个真实的PDF文件
# 为了演示，我们假设 'example.pdf' 存在，并且有文本和图像
# 注意：这里的MockLLM不会真的处理图像，只是模拟调用行为
# 如果要运行，请确保您的 example.pdf 存在
# 或者用一个简单的txt文件模拟PDF内容，但图像部分会受限
# 假设 example.pdf 包含以下内容：
# Page 1: "This is a title. Some introductory text about a report. See Figure 1 for details."
# 包含一个图表图像。
# Page 2: "More text about findings. Table 1 summarizes results."
# 包含一些文本和可能通过文本解析识别的表格。
# 为了让代码运行，我们需要一个虚拟的PDF文件或者跳过PDF加载部分
# 这里我们创建一个模拟的PDFProcessor，假设它能返回一些预设内容
class MockPDFProcessor(PDFProcessor):
def __init__(self):
super().__init__("mock_path.pdf")
self.doc = True # 模拟已加载
def get_page_info(self, page_number: int) -> Dict[str, Any]:
if page_number == 0:
# 模拟第一页，包含文本和图像
return {
"page_number": 0,
"text": "This is the first page of the annual report. It contains an executive summary and a key performance indicator chart. Figure 1 shows the revenue trend over the last five years. Please analyze this chart carefully to understand the growth trajectory.",
"images": [{
"bbox": (100, 200, 500, 400),
"object_type": "figure",
"width": 400, "height": 200,
"stream": type('obj', (object,), {'get_data': lambda self: b'mock_image_data'})() # 模拟图像流
}],
"tables": []
}
elif page_number == 1:
# 模拟第二页，主要是文本和一些通过文本识别的表格
return {
"page_number": 1,
"text": "Page two discusses operational details. Employee satisfaction remained high. Table 1 below details the regional sales figures for Q1. No complex charts on this page, mostly descriptive text and a simple data table.",
"images": [],
"tables": [{
"bbox": (50, 500, 550, 600),
"text_content": "Region | Sales Q1n-------|---------nNorth | 1.2MnSouth | 0.8M"
}]
}
else:
return {}
def close_pdf(self):
print("Mock PDF closed.")
mock_pdf_processor = MockPDFProcessor()
mock_llm = MockLLM() # 用于代理决策和文本工具
mock_vision_llm = MockLLM() # 用于视觉工具
parser_agent = AgenticParser(mock_pdf_processor, mock_llm, mock_vision_llm)
parsed_results = parser_agent.parse_document()
print("n--- 最终解析报告 ---")
for page_result in parsed_results:
print(f"Page {page_result['page_number'] + 1}:")
print(f" Text Summary: {page_result['text_summary']}")
print(f" Visual Insights: {page_result['visual_insights']}")
print(f" Agent Reflection: {page_result.get('agent_reflection', '无')}")
print("-" * 30)
代码解释：
MockLLM: 这是一个模拟的LLM，它根据输入信息简单地模拟返回工具调用或文本回复。在实际项目中，这里会替换为调用OpenAI、Anthropic、Google等厂商的API。Tool抽象类: 定义了所有工具的通用接口，包括名称、描述和参数（遵循JSON Schema）。TextInfoExtractionTool: 模拟一个纯文本处理工具，它会接收文本和分析指令，然后让LLM对文本进行摘要或信息提取。VisualAnalysisTool: 模拟一个视觉分析工具。它接收图像的base64编码和分析指令。在_execute_tool_call中，我们演示了如何将pdfplumber提取的图像流转换为base64字符串。这个工具的run方法内部会调用一个多模态LLM（或专门的CV模型）来处理图像。AgenticParser: 这是我们的核心代理类。__init__：初始化PDF处理器、LLM、视觉LLM和工具。_get_llm_system_prompt：定义了代理的系统级指令，引导其行为。_prepare_page_observation：将当前页面的文本、图像区域信息和历史记忆整合，作为LLM的输入。_execute_tool_call：根据LLM的tool_calls指令，实例化并运行相应的工具。特别处理了visual_analysis_tool，将图像数据编码后传递。parse_document：主循环，逐页迭代。在每一页中，代理首先观察，然后LLM进行决策（是调用文本工具还是视觉工具，或者直接回复）。工具执行后，代理会更新其内存和最终输出。
关键的决策逻辑（在 MockLLM.chat 和 _prepare_page_observation 中体现）：
- 观察: 代理的观察输入包含了页面的文本内容和图像区域的元数据（如边界框）。LLM通过分析这些信息来形成对页面的初步理解。
- 判断: LLM会根据其训练知识和系统提示，判断页面的关键信息是否包含在图像中。例如，如果页面文本中提到“图表显示了增长趋势”，且页面确实检测到图像，LLM更有可能决定调用
visual_analysis_tool。 - 工具选择与参数生成: LLM不仅选择工具，还会生成调用该工具所需的参数（例如，对视觉模型来说，就是具体的
analysis_prompt）。
C. 上下文管理与记忆
一个智能代理的“智能”很大程度上取决于它如何管理上下文和记忆。
- 短期记忆（Short-term Memory）: 指的是当前对话轮次中的信息，例如当前页的观察结果、LLM的思考过程、工具的输出。这些信息直接参与到当前页的决策。在我们的
parse_document方法中，messages列表就是短期的对话历史。 - 长期记忆（Long-term Memory）: 指的是代理在处理整个文档过程中积累的知识。这包括之前页面的解析结果、文档的整体结构、关键实体和关系等。长期记忆通常通过向量数据库实现检索增强生成 (RAG)，或者简单地存储在结构化数据中供LLM查询。
在我们的示例中，self.memory是一个简单的列表，用于存储每页的摘要发现。在_prepare_page_observation中，我们从这个记忆中检索最近几页的信息，作为当前页LLM决策的额外上下文。
# AgenticParser._prepare_page_observation 方法中的记忆检索片段
if self.memory:
observation_str += "n--- 历史解析发现 (最近3页) ---n"
for item in self.memory[-3:]: # 只看最近3页
observation_str += f" - 页 {item['page_number'] + 1}: {item['summary_of_findings']}n"这种简单的记忆方式对于小文档或相邻页面强相关的情况有效。对于大型复杂文档，更高级的RAG系统会根据当前查询（即当前页的观察）从整个文档的知识库中检索最相关的历史信息。
D. 错误处理与鲁棒性
任何实际系统都必须考虑错误。在Agentic Document Parsing中，潜在的错误点很多：
- PDF文件损坏或格式异常。
- OCR识别错误，导致文本不完整或不准确。
- 图像质量差，视觉模型无法准确解析。
- LLM推理失败或生成无效的工具调用参数。
- API调用超时或速率限制。
提高鲁棒性的策略：
- 异常捕获: 在每个关键步骤（PDF加载、工具调用、LLM推理）中加入
try-except块。 - 重试机制: 对于临时的API错误或网络问题，可以实现指数退避重试。
- 默认/回退策略: 如果视觉模型解析失败，代理可以尝试仅依赖文本信息，或者标记该区域为“无法解析”。
- 人工干预点: 对于高置信度低的解析结果，或特定类型的错误，可以设计一个机制，将问题提交给人工审核。
- 日志记录: 详细记录代理的决策过程、工具输入输出和任何错误，方便调试和优化。
四、 进阶考量与未来展望
Agentic Document Parsing的潜力远不止于此。
- 更智能的图像区域识别与分类: 目前我们依赖
pdfplumber识别通用图像区域。未来可以集成更专业的计算机视觉模型，在预处理阶段就能识别出“这是一个柱状图”、“这是一个流程图”、“这是一个签名区域”，并提供更精细的边界框和初步分类信息，从而为代理的决策提供更丰富的上下文。 - 多模态原生代理: 随着多模态LLM的发展，代理本身可能不再需要显式地“调用”一个独立的视觉模型工具。未来的LLM可以直接接收文本和图像作为输入，并直接在内部进行多模态推理和决策，从而简化了工具编排的复杂性。
- 自适应学习与反馈循环: 代理可以从每次解析的成功与失败中学习。例如，通过用户对解析结果的反馈，代理可以调整其决策权重、优化工具调用的提示词，甚至更新其内部知识表示。这可以引入强化学习或在线学习的机制。
- 知识图谱构建: 将解析出的实体、属性和关系存储在一个知识图谱中。这不仅能提供更强大的查询能力，还能帮助代理在后续解析中进行更复杂的推理，例如，如果它知道“公司A”的CEO是谁，当在另一份文档中看到该CEO的名字时，能更快地将其与公司A关联起来。
- 交互式解析: 允许用户在解析过程中与代理进行交互，提供指导、纠正错误或提出特定的解析目标，使系统更具灵活性和可用性。
- 性能与成本优化: 视觉模型的调用通常是昂贵且耗时的。智能代理需要权衡解析的深度和成本。例如，对于非核心的图表，可能只需要一个粗略的描述，而不是详尽的数据提取。可以通过动态调整视觉模型的精度或调用策略来优化。
五、 智能文档解析的新纪元
我们今天探讨的Agentic Document Parsing，标志着文档智能处理从基于规则、模板的“硬编码”时代，向着基于智能代理、自主决策的“软智能”时代迈进。通过赋予系统观察、规划、行动和反思的能力，我们能够构建出更具弹性、更通用的文档理解系统。它不再仅仅是信息的提取者，更是意义的发现者。这无疑将为企业的数据分析、知识管理、自动化流程等领域带来革命性的变革，开启一个智能文档处理的新纪元。