知识图谱怎么喂给多智能体？

Aemond

372人浏览 · 2026-04-19 19:07:08

Aemond · 2026-04-19 19:07:08 发布

上一篇博客里我解释了为什么最终选择 Multi-Agent 架构而不是 Single-Agent：用药安全审查天然需要多专业协作，临床药师看相互作用、内科主治看适应证、过敏专员看过敏史、药房库管看库存——各司其职，协同会诊。

理论上这个决策是对的。

但这周真正开始搭多智能体会诊链路的时候，我才发现一个被我忽略的问题：**五个智能体共享同一套知识图谱，但每个智能体对图谱数据的需求不同、消费方式也不同**。临床药师需要看药物-药物之间的关系，过敏专员需要看药物-过敏原之间的关系，内科主治需要看药物-适应证之间的关系——同一张图，不同视角。

图谱返回的是结构化数据，但每个智能体消费的是自然语言，而且**每个智能体需要看到的重点不一样**。中间这一层怎么设计，文章几乎没有，开源代码也几乎没有参考。

这篇博客就是记录我在解决这个问题过程中的设计决策和踩的几个坑。

一、检索模块的核心逻辑——多智能体视角的改造

图谱检索的目标是：给定一组实体（药物、患者状态、食物、过敏原），找出它们之间所有存在的冲突边。

但在多智能体场景下，检索逻辑需要做一次**分诊**——不是把所有冲突都返回给同一个 LLM，而是按智能体类型分别投递：

```python
# graph/retriever.py
from graph.knowledge_graph import GRAPH
from typing import Optional

def build_alias_index(graph: dict) -> dict:
"""构建别名到标准节点ID的反向索引"""
alias_index = {}
for node_id, node_data in graph["nodes"].items():
alias_index[node_id] = node_id # 标准名本身也加进去
for alias in node_data.get("aliases", []):
alias_index[alias] = node_id
return alias_index

ALIAS_INDEX = build_alias_index(GRAPH)

def normalize_entity(name: str) -> Optional[str]:
"""把输入名称映射到标准节点ID，找不到返回 None"""
return ALIAS_INDEX.get(name, None)

def search_conflicts_by_agent(entities: list[str], agent_type: str) -> dict:
"""
给定一组实体名和智能体类型，检索该智能体需要关注的冲突边

返回格式：
{
"conflicts": [...],
"unrecognized": [...],
"status": "found" | "not_found" | "partial"
}
"""
# 第一步：实体归一化
normalized = []
unrecognized = []
for entity in entities:
std_id = normalize_entity(entity)
if std_id:
normalized.append(std_id)
else:
unrecognized.append(entity)

if not normalized:
return {"conflicts": [], "unrecognized": unrecognized, "status": "not_found"}

# 第二步：路径遍历，找到涉及这些节点的边，并按 affected_agents 过滤
entity_set = set(normalized)
conflicts = []
for edge in GRAPH["edges"]:
# 检查这条边是否涉及当前实体集合中的节点
if edge["source"] in entity_set or edge["target"] in entity_set:
other = edge["target"] if edge["source"] in entity_set else edge["source"]
other_node = GRAPH["nodes"].get(other, {})

# 患者状态、食物、过敏原节点不需要被"输入"，只要图谱里有关系就报出来
if other in entity_set or other_node.get("type") in ("condition", "food", "allergen"):
# 关键：按智能体类型过滤
affected = edge.get("affected_agents", [])
if agent_type in affected or "all" in affected:
conflicts.append({
"source": edge["source"],
"target": edge["target"],
"relation": edge["relation"],
"severity": edge["severity"],
"mechanism": edge["mechanism"],
"recommendation": edge["recommendation"]
})

status = "found" if conflicts else "not_found"
if unrecognized:
status = "partial"

return {
"conflicts": conflicts,
"unrecognized": unrecognized,
"status": status
}
```

逻辑上不复杂：建别名索引 → 归一化输入实体 → 遍历所有边找命中 → **按智能体类型过滤** → 返回结构化结果。时间复杂度 O(V + E)，对于我们目前这个规模的知识图谱完全够用。

这个改造的核心价值在于：**同一个检索函数，不同的智能体调用会得到不同的结果**。临床药师调用时看到的是 DDI 和 CONTRAINDICATION 类的冲突，过敏专员调用时看到的是 CROSS_ALLERGY 类的冲突，内科主治调用时看到的是适应证相关的冲突。每个智能体只拿到自己需要的数据，不会被无关信息干扰。

二、最大的问题：图谱结果怎么分别序列化给不同的智能体

检索逻辑写完之后，我以为接下来很简单——把结果 JSON 分别塞进各个智能体的 Prompt 就好了。

然后我测试了一下，发现效果很差。

测试用例：患者同时服用布洛芬和华法林，有胃溃疡史，自述对青霉素过敏。

图谱返回了三条冲突边：布洛芬-华法林的出血风险、布洛芬-胃溃疡的黏膜损伤、布洛芬-青霉素的交叉过敏信息。

如果我把原始 JSON 原封不动地塞给**每一个**智能体：

```json
{"conflicts": [{"source": "ibuprofen", "target": "warfarin", "relation": "DDI", ...}, {"source": "ibuprofen", "target": "gastric_ulcer", ...}, {"source": "ibuprofen", "target": "penicillin", ...}]}
```

LLM 确实能读懂，但出现了三个新问题：

**问题一：信息过载。** 过敏专员收到了药物相互作用的数据，但它根本不需要看这些。它的职责是审查过敏史和交叉过敏，DDI 数据对它来说是噪音，反而可能干扰判断。

**问题二：严重等级的语义没有被强化。** `"severity": "HIGH"` 对 LLM 来说只是一个字符串，它不一定知道 HIGH 意味着“必须阻断，不能有任何模糊措辞”。不同智能体对“HIGH”的理解应该是一致的，但如果每个智能体的 Prompt 里都重复定义一遍，维护成本太高。

**问题三：节点 ID 是英文的，LLM 在生成中文解释时有时候会夹杂英文节点名，读起来很奇怪。**

所以我加了一个**多智能体序列化层**，专门把图谱返回的结构化数据翻译成**每个智能体专属的**自然语言格式：

```python
# graph/serializer.py
SEVERITY_PROMPT_MAP = {
"HIGH": "〖高危 · 必须阻断〗",
"MEDIUM": "〖中危 · 建议调整〗",
"LOW": "〖低危 · 注意提示〗",
}

RELATION_DESC_MAP = {
"DDI": "药物相互作用",
"CONTRAINDICATION": "用药禁忌",
"FOOD_DRUG": "药食冲突",
"CROSS_ALLERGY": "交叉过敏",
}

# 每个智能体关注的关系类型
AGENT_RELATION_FILTER = {
"clinical_pharmacologist": ["DDI", "CONTRAINDICATION", "FOOD_DRUG"],
"allergy_specialist": ["CROSS_ALLERGY"],
"internal_medicine": ["CONTRAINDICATION", "INDICATION"],
"pharmacy_inventory": ["SUBSTITUTION", "SHORTAGE"],
}

def serialize_for_agent(retrieval_result: dict, agent_type: str) -> str:
"""
把图谱检索结果序列化为面向特定智能体的自然语言描述
"""
if not retrieval_result["conflicts"]:
if retrieval_result["unrecognized"]:
return (
f"知识图谱检索完成。以下实体未能识别，无法判断其安全性："
f"{', '.join(retrieval_result['unrecognized'])}。"
f"请在最终回复中标注这些实体存在信息盲区。"
)
return "知识图谱检索完成，未发现该智能体需要关注的已知冲突。"

# 按智能体类型过滤关系类型
allowed_relations = AGENT_RELATION_FILTER.get(agent_type, [])
filtered_conflicts = [
c for c in retrieval_result["conflicts"]
if c["relation"] in allowed_relations
]

if not filtered_conflicts:
return "知识图谱检索完成，未发现该智能体需要关注的已知冲突。"

lines = [f"知识图谱检索发现以下冲突（{agent_type} 专供），请严格按照风险等级作出决策：\n"]

for i, conflict in enumerate(filtered_conflicts, 1):
severity_label = SEVERITY_PROMPT_MAP.get(conflict["severity"], "〖未知等级〗")
relation_desc = RELATION_DESC_MAP.get(conflict["relation"], conflict["relation"])

src_node = GRAPH["nodes"].get(conflict["source"], {})
tgt_node = GRAPH["nodes"].get(conflict["target"], {})
src_name = src_node.get("aliases", [conflict["source"]])[0]
tgt_name = tgt_node.get("aliases", [conflict["target"]])[0]

lines.append(
f"冲突 {i}：{severity_label}\n"
f" 类型：{relation_desc}\n"
f" 涉及：{src_name} ↔ {tgt_name}\n"
f" 机制：{conflict['mechanism']}\n"
f" 建议：{conflict['recommendation']}\n"
)

if retrieval_result["unrecognized"]:
lines.append(
f"\n注意：以下实体未在知识图谱中找到记录，存在信息盲区："
f"{', '.join(retrieval_result['unrecognized'])}。"
)

lines.append(
"\n决策规则：存在任意〖高危·必须阻断〗冲突时，最终结论必须为 blocked，"
"不得使用模糊或建议性措辞。"
)
return "\n".join(lines)
```