企业Agent实战搭建全流程2/3——药企临床方案长文档处理
·
这是"药企AI Agent落地三部曲"的第2篇。三篇分别解决：
-
第1篇：文献调研与竞品情报——信息淹没，人工整理效率极低
-
本篇：临床方案与注册文件——长文档修改联动，格式校验耗时
-
第3篇：跨部门数据孤岛——多部门数据不通，协作靠手工转格式
本三部曲完整记录了从单Agent逐步扩展到多Agent协作系统的过程。所有工具全线开源，可私有化部署。
上篇用Dify + PGVector + Mem0 + Qwen3.5搭了一套竞品情报Agent，周报生成从4小时压缩到30-40分钟。本篇在既有的架构基础上，增加长文档理解和参数联动能力，解决临床方案和注册文件的修改痛点。
1. 问题描述
药企注册部和医学写作的人应该都有经历：一份临床方案动辄120页，CDE发回一个审评意见说"入组标准需要调整"，看起来就改一段话的事儿，实际上改完入组标准，样本量要重新算，统计方法可能要跟着调，安全性章节里关于受试者安全监测的内容要更新，知情同意书里关于入排标准的描述也要同步修改。
改一个地方，牵连五个章节。这就是临床方案修改的日常。
几个数字比较说明问题：
-
一份临床方案从初稿到提交，平均要改5-7轮才定稿
-
每轮修改涉及3-5个章节的联动更新
-
手动检查章节之间的交叉引用一致性，平均每轮能发现8-12处遗漏或不一致的地方
-
ICH CTD格式规范有几百条规则，手动把整份文件过一遍要2天
-
提交到CDE之后被打回来的格式问题，平均47个。是的，你没看错，47个
这些问题的本质不是"写不出来"，而是"改不动"。一个120页的文档，牵一发而动全身，人脑记不住哪些地方互相依赖，只能靠反复通读来检查。
2. 方案设计
核心思路先说清楚：长文档处理的难点不是LLM能不能读——Qwen3.5支持32K上下文，120页PDF大概40-50万字符，确实塞得进去。但是塞进去之后呢？让它"记住"120页的内容然后回答问题，准确率会断崖式下降。
真正难的是两件事：结构理解和关联追踪。
结构理解是说，Agent要知道这份文档的骨架是什么样的——哪些是章节、哪些是表格、哪些是公式，它们之间的层级关系是什么。关联追踪是说，当入组标准里的"年龄下限从18改成16"这个参数发生变化的时候，Agent要能知道哪些下游的段落、表格、公式会受到影响。
所以我们要做的不是做文档编辑器，是给Agent加上"读懂长文档"和"追踪参数影响"的能力。
新增技术选型
基础架构可以回到前面查看第1篇，这里只说新增的部分。
|
新增组件 |
选型 |
为什么选它 |
|---|---|---|
|
长文档解析 |
MinerU章节树模式 |
复用MinerU，切换到章节级解析模式，不用额外装新东西 |
|
参数依赖图谱 |
NetworkX |
开源图计算库，纯Python，轻量级，适合参数间依赖关系建模 |
|
格式校验 |
Pydantic |
结构化验证框架，把ICH CTD规范翻译成可执行规则 |
|
差异标注 |
difflib + python-docx |
Python标准库加Word修订标记，用户逐条接受/拒绝 |
|
知识库扩展 |
PGVector新增表 |
复用已有PGVector，新增临床方案模板库和注册规范库 |
|
记忆层扩展 |
Mem0任务记忆 |
复用已有Mem0，新增"改了什么→影响了什么"的链路记忆 |
架构图如下：
┌──────────────────────────────────────────┐
│ 单Agent · 文档双引擎（Dify） │
│ ┌────────────┐ ┌──────────┐ ┌────────┐ │
│ │章节树引擎 │ │参数依赖图谱│ │格式校验 │ │
│ │MinerU │ │NetworkX │ │Pydantic│ │
│ └────────────┘ └──────────┘ └────────┘ │
│ ┌────────────┐ ┌──────────────────────┐ │
│ │差异标注 │ │知识库（章节级切片） │ │
│ │difflib │ │PGVector + Mem0 │ │
│ └────────────┘ └──────────────────────┘ │
│ LLM: Qwen3.5-35B-A3B (私有化部署) │
│ 复用: RustFS文档存储 + BGE-Reranker │
└──────────────────────────────────────────┘新增组件都是纯Python库，没有额外的GPU消耗，部署成本很低。下面按Step拆解实施过程。
3. 实施过程
Step 1：长文档章节树构建（Day 1-3）
这一步要做的事情是：把一份120页的临床方案PDF，解析成有层级结构的章节树，然后按章节切片存入PGVector。
为什么要做章节树？因为临床方案不是散文，它有非常明确的结构。ICH CTD格式的临床方案，从模块1到模块5，每个模块下面有固定的一级标题、二级标题、三级标题，层级关系是确定的。如果只是把PDF拆成一页一页的文本片段再向量化，检索的时候只能做关键词匹配，找不到"这份方案里所有跟样本量相关的段落"这种结构化信息。
先装依赖：
# MinerU的安装在第1篇已经安装过了，这里确认一下版本pip show magic-pdf# 如果还没装，执行这个pip install magic-pdf[full]
章节树构建的完整代码：
import json
import re
from magic_pdf.data.data_reader_writer import FileBasedDataWriter, FileBasedDataReader
from magic_pdf.data.dataset import PymuDocDataset
def build_chapter_tree(pdf_path, output_path):
"""
把临床方案PDF解析成章节树结构。
输出是一个嵌套的JSON，每个节点包含：
- title: 章节标题
- level: 层级（1=一级标题，2=二级标题...）
- content: 该章节的文本内容（不包含子章节）
- children: 子章节列表
- page_range: 起始页和结束页
"""
# 第一步：用MinerU的章节树模式解析PDF
# 关键是开启layout analysis，这样才能准确识别标题层级
reader = FileBasedDataReader('')
pdf_bytes = reader.read(pdf_path)
ds = PymuDocDataset(pdf_bytes)
# MinerU的章节结构识别需要开启layout analysis模式
# 配置参数在parse_params里指定
parse_result = ds.apply(
parse_config={
'parse_method': 'auto', # 自动检测排版类型
'layout_analysis': True, # 开启布局分析（关键！）
'formula_analysis': True, # 公式识别（临床方案里有统计公式）
'table_analysis': True, # 表格识别（方案里有大量表格）
}
)
md_content = parse_result.get_markdown()
# 第二步：从Markdown中提取章节树
# 临床方案的标题有明确的Markdown格式：# 一级标题、## 二级标题、### 三级标题
chapter_tree = parse_markdown_to_tree(md_content)
# 第三步：保存章节树JSON
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(chapter_tree, f, ensure_ascii=False, indent=2)
return chapter_tree
def parse_markdown_to_tree(md_content):
"""
把Markdown文本解析成章节树。
思路：按行扫描，遇到标题行就创建新节点，
标题之间的内容属于当前节点，嵌套关系靠标题层级决定。
"""
lines = md_content.split('\n')
root = {
'title': 'root',
'level': 0,
'content': '',
'children': [],
'page_range': [1, 1]
}
stack = [root] # 用栈来追踪当前的父节点
current_content = []
# ICH CTD格式的章节编号正则，用来辅助识别层级
# 比如 "1.1" "2.3.1" 这种编号
numbering_pattern = re.compile(r'^(#{1,6})\s+(?:(\d+(?:\.\d+)*)\s+)?(.+)')
for line in lines:
match = numbering_pattern.match(line.strip())
if match:
# 遇到标题行，先把之前累积的内容保存到当前节点
if current_content:
stack[-1]['content'] = '\n'.join(current_content).strip()
current_content = []
# 创建新节点
level = len(match.group(1)) # # 的个数就是层级
number = match.group(2) or '' # 章节编号，可能为空
title = match.group(3).strip()
new_node = {
'title': title,
'level': level,
'number': number,
'content': '',
'children': [],
'page_range': [0, 0]
}
# 找到合适的父节点：栈里最后一个level小于当前level的节点
while len(stack) > 1 and stack[-1]['level'] >= level:
stack.pop()
stack[-1]['children'].append(new_node)
stack.append(new_node)
else:
current_content.append(line)
# 别忘了最后一段内容
if current_content:
stack[-1]['content'] = '\n'.join(current_content).strip()
return root
def flatten_chapters_to_pgvector(chapter_tree, pg_connection, doc_id):
"""
把章节树拍平，每个章节作为一个独立片段存入PGVector。
metadata里带上章节编号、标题、层级、所属文档ID，
这样检索的时候可以做结构化过滤。
"""
import psycopg2
from pgvector.psycopg2 import register_vector
conn = psycopg2.connect(pg_connection)
register_vector(conn)
cur = conn.cursor()
# 确保表存在（新增一张章节级的表）
cur.execute("""
CREATE TABLE IF NOT EXISTS chapter_embeddings (
id SERIAL PRIMARY KEY,
doc_id TEXT,
chapter_number TEXT,
chapter_title TEXT,
chapter_level INTEGER,
parent_number TEXT,
content TEXT,
page_start INTEGER,
page_end INTEGER,
embedding vector(1024)
)
""")
conn.commit()
# 递归遍历章节树
def walk(node, parent_number=''):
if node['level'] == 0:
# root节点不入库
for child in node['children']:
walk(child, '')
return
chapter_number = node.get('number', '')
content = node['content']
if content.strip():
# 用embedding模型生成向量（继续使用bge-m3）
# 实际部署时通过HTTP调用embedding服务
embedding = get_embedding(content[:8000]) # 截断超长内容
cur.execute("""
INSERT INTO chapter_embeddings
(doc_id, chapter_number, chapter_title, chapter_level,
parent_number, content, page_start, page_end, embedding)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (
doc_id, chapter_number, node['title'], node['level'],
parent_number, content,
node['page_range'][0], node['page_range'][1],
embedding
))
for child in node['children']:
walk(child, chapter_number)
walk(chapter_tree)
conn.commit()
cur.close()
conn.close()
def get_embedding(text):
"""
调用embedding服务生成向量。
使用bge-m3 embedding服务。
"""
import requests
response = requests.post(
'http://内网IP:8001/v1/embeddings',
json={
'model': 'bge-m3',
'input': text
}
)
return response.json()['data'][0]['embedding']运行方式：
# 解析一份120页的临床方案
tree = build_chapter_tree(
pdf_path='/data/protocols/PROT-2024-001_v3.pdf',
output_path='/data/protocols/PROT-2024-001_chapter_tree.json'
)
# 存入PGVector
flatten_chapters_to_pgvector(
chapter_tree=tree,
pg_connection='postgresql://user:pass@内网IP:5432/pharma_db',
doc_id='PROT-2024-001'
)踩坑记录
坑1：120页PDF一次性丢给LLM直接超窗口。第一版方案是"把PDF全文塞进Qwen3.5让它理解结构"，结果120页PDF大概40-50万字符，远远超过32K上下文。就算能塞进去，LLM对长文本中间部分的理解也会严重衰减。必须分层解析：先用MinerU提取目录结构和章节骨架，再逐章节送进LLM做深度理解。MinerU负责"看骨架"，LLM负责"读内容"，分工明确。
坑2：临床方案有固定范式，先定义模板再匹配，比纯靠LLM理解准得多。ICH CTD格式的临床方案结构是高度标准化的——模块2是临床概述、模块3是药学、模块5是临床研究报告。如果你先定义一套模板（"模块5下面一定有5.3.1研究设计、5.3.2入组标准……"），然后让MinerU去匹配这个模板，识别准确率比"让LLM自己猜这段是什么标题"要高得多。模板可以覆盖80%的结构，剩下20%的非标章节再交给LLM处理。
官方文档：[MinerU](https://github.com/opendatalab/MinerU)
Step 2：参数联动关系图谱（Day 4-7）
章节树解决的是"找到相关内容在哪"的问题。但是光找到还不够，你还得知道参数之间的依赖关系。
举个例子：入组标准里的"年龄下限"从18改成16，那哪些地方要跟着改？至少这些：
-
样本量计算（入组人群范围变了，预估的入组率会变）
-
统计方法（分层因素可能要加一个年龄段）
-
安全性章节（未成年受试者的额外安全监测要求）
-
知情同意书（年龄相关描述）
-
研究设计流程图里的筛选/入组部分
这种"改A会影响B、C、D"的关系，用图来建模最合适。NetworkX就是干这个的——构建一个有向图，节点是参数，边是依赖关系，给一个修改点，BFS遍历一遍就能拿到所有受影响的下游节点。
先装依赖：
pip install networkx完整的参数依赖图谱构建和查询代码：
import json
import networkx
from collections import deque
class ParameterDependencyGraph:
"""
参数依赖关系图谱。
用NetworkX的有向图来建模参数之间的依赖关系。
方向：A -> B 表示"修改A会影响B"。
"""
def __init__(self, rules_file=None):
"""
初始化依赖图谱。
rules_file: JSON格式的依赖规则文件路径。
不同临床方案的参数关系不完全一样，所以规则做成可配置的JSON文件，
不写死在代码里。
"""
self.graph = nx.DiGraph()
self.param_metadata = {} # 存节点的额外信息，比如参数在文档中的位置
if rules_file:
self.load_from_json(rules_file)
def load_from_json(self, rules_file):
"""
从JSON配置文件加载依赖规则。
JSON格式示例见下方。
"""
with open(rules_file, 'r', encoding='utf-8') as f:
rules = json.load(f)
# 添加节点
for param in rules.get('parameters', []):
self.graph.add_node(
param['id'],
name=param['name'],
module=param['module'], # 所属章节模块
chapter=param.get('chapter', ''), # 章节编号
description=param.get('description', '')
)
self.param_metadata[param['id']] = param
# 添加依赖边
for dep in rules.get('dependencies', []):
self.graph.add_edge(
dep['from'], # 被修改的参数
dep['to'], # 受影响的参数
reason=dep.get('reason', ''), # 为什么会有依赖关系
strength=dep.get('strength', 'medium') # 影响程度：high/medium/low
)
def get_impact_scope(self, changed_param_id, max_depth=5):
"""
给定一个被修改的参数，返回所有受影响的下游参数。
用BFS遍历，最多遍历max_depth层。
返回列表，每项包含：参数ID、名称、影响路径、影响程度。
"""
if changed_param_id not in self.graph:
return []
result = []
visited = set()
queue = deque()
# 初始种子：直接下游节点
for successor in self.graph.successors(changed_param_id):
edge_data = self.graph.edges[changed_param_id, successor]
queue.append({
'param_id': successor,
'path': [changed_param_id, successor],
'depth': 1,
'reason': edge_data.get('reason', ''),
'strength': edge_data.get('strength', 'medium')
})
while queue:
item = queue.popleft()
param_id = item['param_id']
if param_id in visited or item['depth'] > max_depth:
continue
visited.add(param_id)
node_data = self.graph.nodes[param_id]
result.append({
'param_id': param_id,
'param_name': node_data.get('name', ''),
'module': node_data.get('module', ''),
'chapter': node_data.get('chapter', ''),
'path': item['path'],
'reason': item['reason'],
'strength': item['strength'],
'depth': item['depth']
})
# 继续遍历下游
for successor in self.graph.successors(param_id):
if successor not in visited:
edge_data = self.graph.edges[param_id, successor]
queue.append({
'param_id': successor,
'path': item['path'] + [successor],
'depth': item['depth'] + 1,
'reason': edge_data.get('reason', ''),
'strength': edge_data.get('strength', 'medium')
})
# 按影响程度排序（high > medium > low）
strength_order = {'high': 0, 'medium': 1, 'low': 2}
result.sort(key=lambda x: strength_order.get(x['strength'], 3))
return result
def get_subgraph_by_module(self, module_name):
"""
按模块拆分子图，方便可视化。
比如只看"入组标准"模块相关的参数和依赖关系。
"""
nodes_in_module = [
n for n, data in self.graph.nodes(data=True)
if data.get('module') == module_name
]
# 包含模块内节点，以及它们之间的边
return self.graph.subgraph(nodes_in_module)
def visualize_impact(self, changed_param_id, output_path):
"""
生成影响范围的可视化图片。
按模块着色，方便一眼看出影响波及了哪些章节。
"""
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
impact = self.get_impact_scope(changed_param_id)
if not impact:
return
# 收集涉及的模块
involved_modules = set()
involved_modules.add(self.graph.nodes[changed_param_id].get('module', ''))
for item in impact:
involved_modules.add(item['module'])
# 构建子图
involved_nodes = {changed_param_id}
for item in impact:
involved_nodes.add(item['param_id'])
subgraph = self.graph.subgraph(involved_nodes)
# 颜色映射
colors = plt.cm.Set3(range(len(involved_modules)))
module_color_map = dict(zip(involved_modules, colors))
node_colors = []
for node in subgraph.nodes():
module = subgraph.nodes[node].get('module', '')
node_colors.append(module_color_map.get(module, 'gray'))
plt.figure(figsize=(14, 10))
pos = nx.spring_layout(subgraph, k=2, iterations=50)
nx.draw(
subgraph, pos,
node_color=node_colors,
node_size=2000,
with_labels=True,
labels={n: self.graph.nodes[n].get('name', n) for n in subgraph.nodes()},
font_size=8,
arrows=True,
arrowsize=20,
edge_color='#999999'
)
plt.title(f'参数依赖影响范围：{self.graph.nodes[changed_param_id].get("name", changed_param_id)}')
plt.tight_layout()
plt.savefig(output_path, dpi=150)
plt.close()
# ========== 使用示例 ==========
# 加载规则文件（不同方案用不同的规则文件）
dep_graph = ParameterDependencyGraph(
rules_file='/data/rules/protocol_dependency_rules.json'
)
# 查询：修改"入组年龄下限"会影响哪些参数？
impact = dep_graph.get_impact_scope('inclusion_age_lower')
for item in impact:
print(f"影响: {item['param_name']} ({item['module']}) "
f"| 路径: {' -> '.join(item['path'])} "
f"| 程度: {item['strength']}")
# 生成可视化图
dep_graph.visualize_impact('inclusion_age_lower', '/data/output/impact_age_lower.png')规则文件 protocol_dependency_rules.json 的格式：
{
"parameters": [
{
"id": "inclusion_age_lower",
"name": "入组年龄下限",
"module": "入组标准",
"chapter": "5.3.2",
"description": "受试者入组的最低年龄"
},
{
"id": "sample_size",
"name": "样本量",
"module": "统计设计",
"chapter": "5.3.1",
"description": "计划的入组受试者总数"
},
{
"id": "statistical_method",
"name": "统计方法",
"module": "统计分析",
"chapter": "5.3.3",
"description": "主要分析的统计方法"
},
{
"id": "safety_monitoring",
"name": "安全性监测方案",
"module": "安全性",
"chapter": "5.4",
"description": "受试者安全监测计划"
},
{
"id": "icf_age_description",
"name": "知情同意书年龄描述",
"module": "知情同意",
"chapter": "附录",
"description": "知情同意书中关于年龄要求的描述"
},
{
"id": "randomization_stratification",
"name": "随机化分层因素",
"module": "统计设计",
"chapter": "5.3.1",
"description": "随机化分层的因素列表"
},
{
"id": "enrollment_rate",
"name": "入组速率预估",
"module": "研究进度",
"chapter": "5.3.5",
"description": "每月预计入组的受试者数"
}
],
"dependencies": [
{
"from": "inclusion_age_lower",
"to": "sample_size",
"reason": "入组人群范围变化影响入组速率，从而影响样本量计算",
"strength": "high"
},
{
"from": "inclusion_age_lower",
"to": "safety_monitoring",
"reason": "年龄下限降低意味着可能纳入未成年受试者，需要额外的安全监测",
"strength": "high"
},
{
"from": "inclusion_age_lower",
"to": "icf_age_description",
"reason": "知情同意书中的年龄要求必须与入组标准保持一致",
"strength": "high"
},
{
"from": "inclusion_age_lower",
"to": "enrollment_rate",
"reason": "年龄范围变化影响目标人群规模，进而影响入组速率",
"strength": "medium"
},
{
"from": "sample_size",
"to": "statistical_method",
"reason": "样本量变化可能需要调整统计方法（如从正态近似切换到精确检验）",
"strength": "medium"
},
{
"from": "sample_size",
"to": "randomization_stratification",
"reason": "样本量变化可能影响分层因素的选择和层数",
"strength": "medium"
},
{
"from": "statistical_method",
"to": "sample_size",
"reason": "统计方法变化也可能反过来影响样本量（双向依赖）",
"strength": "medium"
}
]
}这里有一个关键的设计决策要说一下：Agent不是自动改所有关联段落（太危险），而是标记影响范围让人确认。 为什么不做全自动？因为临床方案是提交给监管机构的正式文件，每一处修改都要有人负责。Agent告诉你"改了这里会影响这5个地方"，但具体改不改、怎么改，人来决定。
踩坑记录坑1：第一版把参数依赖关系写死在代码里了。写了一个Python类，每个参数之间的关系用if-else硬编码。结果换了另一份临床方案，参数结构不一样，代码直接不能用了。后来改成JSON配置文件，不同方案用不同的规则文件，代码逻辑完全不用动。这个教训就是：领域知识不要写死在代码里，做成配置。坑2：图太大了，可视化出来一团乱麻。某些复杂的临床方案有200多个参数节点，全画出来根本看不懂。后来改成按模块拆分子图——想看"入组标准"的影响范围就只画这个模块相关的子图。上面的代码里 get_subgraph_by_module 和 visualize_impact 就是干这个的。
官方文档：[NetworkX](https://networkx.org/)
Step 3：ICH CTD格式校验引擎（Day 8-10）
临床方案提交给CDE有严格的格式要求，ICH CTD规范有几百条规则。手动检查一遍要2天，而且人眼检查这种东西效率很低，看久了就麻木了。
这一步用Pydantic把CTD规范翻译成可执行的校验规则。核心思路是：把"规范"变成代码，让机器来检查。
先装依赖：
pip install pydantic完整的格式校验引擎代码：
import re
import json
from typing import List, Optional
from pydantic import BaseModel, field_validator
# ========== 校验结果模型 ==========
class ValidationIssue(BaseModel):
"""单条格式问题的描述"""
location: str # 位置，比如"第3页第2段"
rule_id: str # 规则编号
severity: str # 严重程度：error / warning / info
expected: str # 规范要求什么
actual: str # 当前是什么
suggestion: str # 建议改成什么
class ValidationReport(BaseModel):
"""校验报告"""
doc_id: str
module_checked: str # 检查了哪个CTD模块
total_issues: int
errors: int
warnings: int
issues: List[ValidationIssue]
# ========== 规则定义 ==========
class CTDFormatRules(BaseModel):
"""
ICH CTD格式规则模型。
这里展示5条代表性规则，实际项目里有200+条。
"""
# 规则1：模块标题编号格式
module_title_pattern: str = r'^模块\s*[1-5]\s*：'
# 规则2：章节标题层级——一级标题用"1"，二级标题用"1.1"，三级标题用"1.1.1"
heading_number_pattern: str = r'^\d+(\.\d+)*\s+'
# 规则3：表格必须有编号和标题，格式为"表X-X：标题"
table_caption_pattern: str = r'^表\s*\d+-\d+\s*[：:]\s*.+'
# 规则4：引用文献格式——必须用方括号数字引用，如[1]、[2,3]
citation_pattern: str = r'\[\d+(?:,\s*\d+)*\]'
# 规则5：页边距要求——正文上下边距2.54cm，左右边距3.17cm（Word默认）
# 这条在Word里检查，Markdown阶段只做文本层面的规则
def validate_heading_hierarchy(content, chapter_title):
"""
规则：检查章节标题层级是否连续，不能跳级。
比如二级标题后面不能直接接四级标题。
"""
issues = []
heading_pattern = re.compile(r'^(#{1,6})\s+(?:(\d+(?:\.\d+)*)\s+)?(.+)')
prev_level = 0
for line_no, line in enumerate(content.split('\n'), 1):
match = heading_pattern.match(line.strip())
if not match:
continue
level = len(match.group(1))
number = match.group(2) or ''
# 检查层级是否跳级（比如从2直接跳到4）
if prev_level > 0 and level > prev_level + 1:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='HEADING_001',
severity='error',
expected=f'标题层级应该连续，当前层级{level}的上一级应该是{level-1}',
actual=f'上一个标题层级是{prev_level}，当前跳到{level}',
suggestion=f'在第{line_no}行前补充{level-1}级标题，或调整标题层级'
))
# 检查编号格式是否匹配层级
if number:
parts = number.split('.')
if len(parts) != level:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='HEADING_002',
severity='error',
expected=f'{level}级标题的编号应该有{level}段（如{".".join(["1"]*level)}）',
actual=f'当前编号"{number}"有{len(parts)}段',
suggestion=f'修正编号为{".".join(["1"]*level)}格式'
))
prev_level = level
return issues
def validate_table_captions(content, chapter_title):
"""
规则：每个表格上方必须有编号和标题。
格式为"表X-X：标题"或"表X-X: 标题"。
"""
issues = []
table_caption_pattern = re.compile(r'^表\s*\d+-\d+\s*[：:]\s*.+')
# 检测Markdown表格（以 | 开头的行）
lines = content.split('\n')
in_table = False
table_count = 0
for line_no, line in enumerate(lines, 1):
stripped = line.strip()
if stripped.startswith('|') and not in_table:
# 发现一个新表格
in_table = True
table_count += 1
# 检查表格前面几行有没有标题
has_caption = False
for check_line_no in range(max(0, line_no - 4), line_no - 1):
if check_line_no >= 0 and check_line_no < len(lines):
if table_caption_pattern.match(lines[check_line_no].strip()):
has_caption = True
break
if not has_caption:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='TABLE_001',
severity='error',
expected='表格上方应有"表X-X：标题"格式的编号和标题',
actual='未找到表格标题',
suggestion=f'在第{line_no}行前添加"表{table_count}-1：[表格标题]"'
))
elif not stripped.startswith('|') and in_table:
in_table = False
return issues
def validate_citations(content, chapter_title):
"""
规则：文献引用必须用方括号数字格式[1]、[2,3]。
不能用（张三, 2023）这种APA格式。
"""
issues = []
# 检查是否有APA风格的引用
apa_pattern = re.compile(r'[（(]\s*[A-Z][a-z]+\s*(?:等|et al\.?)?\s*,\s*\d{4}\s*[）)]')
for line_no, line in enumerate(content.split('\n'), 1):
apa_matches = apa_pattern.findall(line)
if apa_matches:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='CITE_001',
severity='error',
expected='使用方括号数字引用格式，如[1]、[2,3]',
actual=f'发现APA格式引用：{", ".join(apa_matches)}',
suggestion='将APA格式引用转换为方括号数字格式'
))
return issues
def validate_numbering_consistency(content, chapter_title):
"""
规则：同一层级的标题编号要连续，不能跳号。
比如"3.1"后面应该是"3.2"，不能直接跳到"3.3"。
"""
issues = []
heading_pattern = re.compile(r'^#{1,6}\s+(\d+(?:\.\d+)*)\s+')
level_counters = {} # 记录每个层级当前的编号
for line_no, line in enumerate(content.split('\n'), 1):
match = heading_pattern.match(line.strip())
if not match:
continue
number = match.group(1)
parts = number.split('.')
level = len(parts)
# 获取父级前缀
if level == 1:
expected_next = level_counters.get('1', 0) + 1
level_counters['1'] = int(parts[0])
if int(parts[0]) != expected_next and level_counters['1'] != 1:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='NUM_001',
severity='warning',
expected=f'一级标题编号应该是{expected_next}',
actual=f'当前编号是{parts[0]}',
suggestion=f'检查是否有遗漏的章节，或修正编号'
))
else:
parent_prefix = '.'.join(parts[:-1])
key = parent_prefix
current_num = int(parts[-1])
expected_next = level_counters.get(key, 0) + 1
level_counters[key] = current_num
if current_num != expected_next and level_counters[key] != 1:
issues.append(ValidationIssue(
location=f'{chapter_title} 第{line_no}行',
rule_id='NUM_002',
severity='warning',
expected=f'{parent_prefix}下的同级标题编号应该连续，当前应该是{expected_next}',
actual=f'当前编号是{parent_prefix}.{current_num}',
suggestion=f'检查是否有遗漏的子章节，或修正编号'
))
return issues
def validate_ctd_module(content, module_name):
"""
对单个CTD模块执行全部校验规则。
返回校验报告。
"""
all_issues = []
# 分模块校验，每个规则独立执行
all_issues.extend(validate_heading_hierarchy(content, module_name))
all_issues.extend(validate_table_captions(content, module_name))
all_issues.extend(validate_citations(content, module_name))
all_issues.extend(validate_numbering_consistency(content, module_name))
# 实际项目里这里还有200+条规则的调用
# 每条规则都是一个独立的检查函数，结构跟上面类似
errors = sum(1 for i in all_issues if i.severity == 'error')
warnings = sum(1 for i in all_issues if i.severity == 'warning')
return ValidationReport(
doc_id=module_name,
module_checked=module_name,
total_issues=len(all_issues),
errors=errors,
warnings=warnings,
issues=all_issues
)
def run_full_validation(chapter_tree_json_path):
"""
对整份文档的所有CTD模块分别执行校验。
分模块出报告，不是一次性检查所有内容。
"""
with open(chapter_tree_json_path, 'r', encoding='utf-8') as f:
tree = json.load(f)
reports = []
# 遍历一级章节（CTD模块级别）
for module_chapter in tree.get('children', []):
module_name = module_chapter.get('title', '未知模块')
content = module_chapter.get('content', '')
# 把子章节的内容也拼进来
def collect_all_content(node):
full = node.get('content', '')
for child in node.get('children', []):
full += '\n\n' + collect_all_content(child)
return full
full_content = collect_all_content(module_chapter)
report = validate_ctd_module(full_content, module_name)
reports.append(report)
print(f"模块 [{module_name}] 校验完成："
f"{report.total_issues} 个问题 "
f"({report.errors} 个错误, {report.warnings} 个警告)")
return reports运行校验：
# 对一份临床方案执行格式校验
reports = run_full_validation('/data/protocols/PROT-2024-001_chapter_tree.json')
# 汇总结果
total = sum(r.total_issues for r in reports)
total_errors = sum(r.errors for r in reports)
print(f"\n全文校验完成：共 {total} 个问题，{total_errors} 个错误需要修正")踩坑记录坑1：规则太多（200+条），Agent一次执行不完。第一版把所有规则塞进一个校验函数，让Agent一次性跑完所有200多条规则。结果执行时间太长，中间还容易超时。后来改成按CTD模块分模块校验——模块2单独出一份报告，模块5单独出一份报告。每个模块的校验时间控制在1-2分钟内，而且用户可以只看自己关心的模块。坑2：Pydantic validator写太复杂会拖慢校验速度。最开始把很多复杂校验逻辑写在Pydantic的validator里，结果一份120页的文档校验要跑10多分钟。后来把复杂校验拆成多个轻量级的检查函数，每个函数只做一件事，然后在主流程里顺序调用。拆分之后单模块校验降到30秒以内。另外，那些不涉及文档内容的纯格式规则（比如页边距、字体大小），放到最后用python-docx直接检查Word文件，不走Pydantic。
官方文档：[Pydantic](https://docs.pydantic.dev/)
Step 4：差异标注与人工确认流程（Day 11-12）
前面三步解决了"找到内容""理解关联""检查格式"的问题。最后一步是：Agent给出修改建议之后，怎么让用户看清楚改了什么，逐条接受或者拒绝。
这一步的设计原则是：所有修改都是"建议"，不自动生效。 用户看到的是带修订标记的Word文档，绿色是新增，红色删除线是删除，跟Word自带的"修订"模式一样。用户逐条审阅，接受的就保留，不接受的就撤销。
先装依赖（python-docx在上一篇中已经装了，difflib是标准库不用装）：
# 确认python-docx已安装
pip show python-docx
# 如果没装
pip install python-docx完整的差异标注和Word输出代码：
import difflib
import json
from docx import Document
from docx.shared import Pt, RGBColor, Inches
from docx.enum.text import WD_COLOR_INDEX
from copy import deepcopy
def generate_paragraph_diff(old_paragraphs, new_paragraphs):
"""
比较旧段落列表和新段落列表的差异。
返回diff结果，每行标记为 added / removed / unchanged。
"""
differ = difflib.SequenceMatcher(
None,
old_paragraphs,
new_paragraphs
)
diff_result = []
for tag, i1, i2, j1, j2 in differ.get_opcodes():
if tag == 'equal':
for idx in range(i1, i2):
diff_result.append({
'status': 'unchanged',
'text': old_paragraphs[idx],
'old_index': idx
})
elif tag == 'replace':
# 替换：先标记删除旧内容，再标记新增新内容
for idx in range(i1, i2):
diff_result.append({
'status': 'removed',
'text': old_paragraphs[idx],
'old_index': idx
})
for idx in range(j1, j2):
diff_result.append({
'status': 'added',
'text': new_paragraphs[idx],
'new_index': idx
})
elif tag == 'delete':
for idx in range(i1, i2):
diff_result.append({
'status': 'removed',
'text': old_paragraphs[idx],
'old_index': idx
})
elif tag == 'insert':
for idx in range(j1, j2):
diff_result.append({
'status': 'added',
'text': new_paragraphs[idx],
'new_index': idx
})
return diff_result
def generate_word_diff(old_doc_path, new_paragraphs, output_path):
"""
生成带修订标记的Word文档。
- 新增的内容：绿色背景
- 删除的内容：红色字体 + 删除线
- 未变的内容：正常显示
"""
# 读取原始文档的段落文本
old_doc = Document(old_doc_path)
old_paragraphs = [p.text for p in old_doc.paragraphs]
# 生成差异
diff_result = generate_paragraph_diff(old_paragraphs, new_paragraphs)
# 创建新文档，写入带标记的内容
new_doc = Document()
# 设置默认样式
style = new_doc.styles['Normal']
style.font.name = '微软雅黑'
style.font.size = Pt(10.5)
for item in diff_result:
para = new_doc.add_paragraph()
if item['status'] == 'unchanged':
# 正常段落
run = para.add_run(item['text'])
run.font.name = '微软雅黑'
run.font.size = Pt(10.5)
elif item['status'] == 'added':
# 新增段落：绿色背景
run = para.add_run(f'[新增] {item["text"]}')
run.font.name = '微软雅黑'
run.font.size = Pt(10.5)
run.font.color.rgb = RGBColor(0, 128, 0) # 绿色
# 给段落加绿色底纹
from docx.oxml.ns import qn
from docx.oxml import OxmlElement
shading = OxmlElement('w:shd')
shading.set(qn('w:fill'), 'E8F5E9') # 浅绿色背景
shading.set(qn('w:val'), 'clear')
para.paragraph_format.element.get_or_add_pPr().append(shading)
elif item['status'] == 'removed':
# 删除段落：红色 + 删除线
run = para.add_run(f'[删除] {item["text"]}')
run.font.name = '微软雅黑'
run.font.size = Pt(10.5)
run.font.color.rgb = RGBColor(200, 0, 0) # 红色
run.font.strike = True # 删除线
new_doc.save(output_path)
return output_path
def generate_change_summary(diff_result, impact_scope):
"""
生成变更摘要报告。
告诉用户：Agent建议了哪些修改，这些修改影响了哪些章节。
"""
summary = {
'total_added': sum(1 for d in diff_result if d['status'] == 'added'),
'total_removed': sum(1 for d in diff_result if d['status'] == 'removed'),
'total_unchanged': sum(1 for d in diff_result if d['status'] == 'unchanged'),
'affected_chapters': [],
'changes': []
}
for item in diff_result:
if item['status'] in ('added', 'removed'):
summary['changes'].append({
'status': item['status'],
'text_preview': item['text'][:100] + ('...' if len(item['text']) > 100 else '')
})
# 关联参数依赖图谱的影响范围
for impact in impact_scope:
summary['affected_chapters'].append({
'chapter': impact.get('chapter', ''),
'param_name': impact.get('param_name', ''),
'module': impact.get('module', ''),
'reason': impact.get('reason', ''),
'strength': impact.get('strength', '')
})
return summary
# ========== 完整流程示例 ==========
def agent_modify_and_review(old_doc_path, modification_instruction, pg_connection):
"""
Agent修改文档并生成差异标注的完整流程。
参数说明：
- old_doc_path: 原始Word文档路径
- modification_instruction: 修改指令，比如"将入组年龄下限从18岁改为16岁"
- pg_connection: PGVector连接字符串
"""
# 第一步：读取原文档段落
old_doc = Document(old_doc_path)
old_paragraphs = [p.text for p in old_doc.paragraphs]
# 第二步：查询参数依赖图谱，获取影响范围
dep_graph = ParameterDependencyGraph(
rules_file='/data/rules/protocol_dependency_rules.json'
)
# 假设修改的是"入组年龄下限"
impact_scope = dep_graph.get_impact_scope('inclusion_age_lower')
print(f"修改影响范围：共 {len(impact_scope)} 个下游参数")
for item in impact_scope:
print(f" - {item['param_name']} ({item['module']}, {item['chapter']})")
# 第三步：调用LLM生成修改建议（这里省略LLM调用，用模拟数据）
# 实际部署时，把修改指令+受影响的章节内容一起发给Qwen3.5
# 让它给出每个受影响章节的修改建议
new_paragraphs = simulate_llm_modification(
old_paragraphs, modification_instruction, impact_scope
)
# 第四步：生成差异
diff_result = generate_paragraph_diff(old_paragraphs, new_paragraphs)
# 第五步：输出带修订标记的Word文档
output_path = old_doc_path.replace('.docx', '_review.docx')
generate_word_diff(old_doc_path, new_paragraphs, output_path)
print(f"审阅文档已生成：{output_path}")
# 第六步：生成变更摘要
summary = generate_change_summary(diff_result, impact_scope)
print(f"变更统计：新增 {summary['total_added']} 段，"
f"删除 {summary['total_removed']} 段，"
f"未变 {summary['total_unchanged']} 段")
return output_path, summary
def simulate_llm_modification(old_paragraphs, instruction, impact_scope):
"""
模拟LLM修改（实际部署时替换为真实的LLM调用）。
这里只用于演示流程。
"""
new_paragraphs = list(old_paragraphs) # 复制一份
# 模拟修改：找到包含"18岁"的段落，改为"16岁"
for i, para in enumerate(new_paragraphs):
if '18岁' in para or '18周岁' in para:
new_paragraphs[i] = para.replace('18岁', '16岁').replace('18周岁', '16周岁')
return new_paragraphs踩坑记录坑1：一开始做成全自动修改，用户不敢提交。第一版的设计是Agent分析完依赖关系后，自动把所有关联段落都改好，直接输出一份新文档。结果注册部的人拿到文档后说"我不知道你改了什么，我不敢用"。信任是慢慢建立的。后来改成差异标注模式——用红色和绿色把所有修改点标出来，用户逐条审阅，接受的留下，不接受的撤销。改了之后用户反馈"现在敢用了"。所以人机协作的关键不是自动化程度有多高，而是信任边界在哪里。坑2：difflib的unifieddiff对Word段落不太好用。difflib默认是做文本行级别的diff，但Word文档的段落不是简单的文本行。第一版直接用unifieddiff对比Word原文和新文本，结果段落对应关系全乱了。解决办法是先把Word段落提取成文本列表，对文本列表做SequenceMatcher级别的diff（不是unifieddiff），然后把diff结果映射回段落，最后用python-docx重新生成带标记的Word文档。核心是不要用unifieddiff，而是用SequenceMatcher自己处理映射逻辑。
官方文档：[difflib](https://docs.python.org/3/library/difflib.html)
4. 运行结果
部署完成之后跑了3轮实际的临床方案修改任务，跟之前纯人工操作做了对比：
|
指标 |
部署前 |
部署后 |
变化 |
|---|---|---|---|
|
临床方案单轮修改时间 |
3天 |
半天~1天 |
↓67-83% |
|
交叉引用遗漏数 |
每轮8-12处 |
0-2处 |
↓80-100% |
|
注册文件格式问题（提交后被退回） |
平均47个 |
3-5个 |
↓89-94% |
|
格式自检耗时 |
2天/次 |
15-20分钟/次 |
↓87-91% |
|
参数联动检查覆盖率 |
人工抽检约60% |
Agent全量检查100% |
↑40% |
以上数据基于3轮实际临床方案修改的统计，取平均值。测试环境与第1篇保持相同：Qwen3.5-35B-A3B私有化部署，双卡A100 40G，内网环境。
增量资源消耗：
-
额外存储：临床方案模板库 + 注册规范库约2GB
-
NetworkX、Pydantic、difflib均为纯Python库，无额外GPU消耗
-
python-docx已有
-
总增量成本很低，主要是规则库的人工维护成本——参数依赖关系的JSON文件和200多条CTD校验规则需要领域专家配合维护，这部分工作量大概需要2-3个人天来初始建立，后续每次方案结构有大的变化时更新一下
5. 总结与下一步
踩坑总结
-
长文档不能直接丢给LLM。必须先用MinerU做章节级拆解，建立章节树结构，然后逐段理解。LLM擅长的是"读懂一段话"，不是"记住120页文档"。把大任务拆成小任务，才是正确的使用方式。
-
参数联动不能写死规则。不同临床方案的结构差异不小，这个方案的入组标准有7个子项，那个方案可能有12个。把依赖关系做成可配置的JSON规则文件，代码逻辑保持不变，换方案只需要改配置文件。
-
格式校验靠规则引擎不靠LLM。ICH CTD的几百条规范，让LLM去"读懂"规范然后检查文档，效果很差——它有时候记得住规则，有时候忘了，输出还不稳定。把规范翻译成Pydantic模型和正则表达式，让代码来检查，结果100%可复现、可审计。
-
人机信任需要设计。差异标注 + 人工确认流程，是用户敢用Agent的关键。全自动不是目标，可控才是。Agent告诉你"改了这里会影响5个地方"，你来决定改不改。这种设计比"Agent全部搞定"要靠谱得多。
核心洞察
长文档处理的关键不是LLM有多强，而是你能不能把领域规范翻译成结构化规则。ICH CTD格式规范、参数之间的依赖关系、章节之间的交叉引用——这些东西都是确定性的知识，完全可以编码成规则和图谱。LLM在这里的角色是"理解语义"（这段在说什么），而不是"记忆规则"（规范是怎么要求的）。
人机协作的关键是信任边界。Agent建议，人确认。不是Agent做得不够好才需要人确认，而是监管文件这种场景天然需要人负责。把这个边界设计好，用户才敢用、才愿意用。
系列预告
第1篇解决了文献情报，第2篇解决了长文档处理。两个Agent各自运转良好。
但是问题来了：医学部的Agent不知道注册部在做什么，注册部的Agent也拿不到临床运营的数据。三个部门的数据像三座孤岛，每次跨部门对接都要手工转格式，3天的活儿。
下一篇，我们用Dify原生的多Agent能力，搭一套跨部门协作系统。
本系列基于真实行业痛点，技术方案基于开源工具和通用架构设计，可迁移至真实场景。文中涉及的工具选型仅代表个人实践偏好，不构成商业推荐。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)