尊敬的各位同仁，
欢迎来到今天的技术讲座。今天我们将深入探讨一个在处理大规模循环图时，尤其是在与现代AI系统（如大型语言模型）结合使用时，日益凸显的挑战——即如何防止重复的背景信息充斥上下文窗口，我们称之为“Contextual De-duplication”，即上下文相关的去重。
一、引言：大规模循环图与上下文窗口的挑战
在复杂的软件系统、知识图谱、社交网络、代码依赖关系，乃至智能体的记忆和决策流程中，我们经常会遇到大规模的循环图结构。这些图拥有海量的节点和边，并且其固有的循环特性意味着从一个节点出发，经过一系列路径，最终可能回到或经过之前访问过的节点。
当我们将这些图中的信息提取出来，作为背景知识输入给一个“上下文窗口”时（例如，一个大型语言模型的输入缓冲区，一个智能体的短期记忆，或一个数据处理管道的临时存储），一个核心问题便浮现出来：如何高效、精确地管理这些信息？
什么是“上下文窗口”？
在本次讲座中，上下文窗口可以被理解为：
- 大型语言模型（LLM）的输入令牌限制： 模型能够同时处理的文本量是有限的，超出部分会被截断或导致性能下降。
- 智能体（Agent）的短期记忆： 智能体在执行任务时需要记住相关信息，但其记忆容量并非无限。
- 分布式系统中的处理批次： 在数据流或图计算中，一次处理的数据量限制。
- 人类认知的焦点： 即使对人类而言，过多的重复信息也会分散注意力，降低理解效率。
为什么重复的背景信息是个问题？
在循环图中，重复的信息尤其常见。例如，一个实体（节点）可能通过多条路径被引用，一个常见的设计模式（子图）可能在代码库中出现多次，或者一个核心概念在知识图谱中被多个相关概念共同指向。
如果不对这些重复信息进行有效管理，会导致：
- 资源浪费： 无论是计算资源（LLM推理成本、CPU/内存）还是存储资源，都在处理和存储冗余数据。
- 效率降低： LLM需要处理更多令牌，导致推理速度变慢。智能体需要筛选更多信息来做出决策。
- 信息淹没： 关键信息可能被大量重复的、低价值的信息所淹没，降低了信噪比。
- 上下文溢出： 这是最直接的后果，尤其是在LLM应用中，重复信息会迅速消耗宝贵的上下文令牌，导致真正重要的信息无法被纳入。
- 认知负担： 对于人类用户而言，阅读包含大量重复内容的报告或输出，会增加理解难度和疲劳感。
因此，“Contextual De-duplication”的核心目标是：在保持信息完整性和准确性的前提下，识别并消除在特定上下文窗口内重复或高度相似的背景信息，从而优化信息密度，提升系统效率和性能。它不仅仅是简单地去重，更强调“上下文相关性”——即同一个信息在不同场景下，其重复的判定标准和处理方式可能不同。
二、问题深度解析：循环图的本质与信息冗余
要有效地进行上下文去重，我们首先需要深入理解循环图的特性以及信息冗余产生的机制。
循环图的本质
一个图G=(V, E)是循环的，如果它包含至少一个循环（Cycle），即从一个节点v出发，通过一系列边和节点，可以回到v的路径。
常见的循环图场景：
- 知识图谱： 实体之间的关系通常是多向且复杂的，例如“A是B的创始人”，“B雇佣了C”，“C与A共同成立了D公司”。这里可能形成A-B-C-A的循环路径。
- 代码依赖图： 模块或函数之间可能存在循环依赖，例如模块A依赖B，B依赖C，C又依赖A。
- 对话流图： 在复杂的对话系统中，用户和AI的交互可能多次回到某个主题或流程节点。
- 推荐系统： 用户-物品-标签-用户之间可能形成复杂的兴趣循环。
信息冗余的来源
在这些循环图中，信息冗余主要来自以下几个方面：
- 路径重复访问： 从起点到目标节点的多个不同路径可能经过相同的中间节点或子图。
- 多源引用： 同一个实体或概念可能被图中的多个不同部分引用或描述。
- 结构相似性： 图中可能存在多个拓扑结构相似的子图，它们描述了相同或高度相似的模式、流程或实体关系。
- 语义相似性： 即使表面文本不同，但信息内容在语义上是等价的。例如，“公司X的CEO是Y”和“Y领导着公司X”。
上下文窗口的限制与信息表示
当我们将图数据转化为文本或其他序列形式输入到上下文窗口时，通常会通过以下方式：
- 节点属性： 节点的名称、描述、类型等。
- 边属性： 边的类型、权重、时间戳等。
- 子图结构： 描述一个小的局部图结构，例如“A与B有关系R，B与C有关系S”。
- 路径描述： 从源到目标的完整路径信息。
这些信息在转化为文本时，如果不对其进行智能管理，很容易导致大量冗余的句子、段落或实体描述。
示例场景：知识图谱中的人物关系
假设我们有一个关于人物和公司的知识图谱。我们要为LLM提供关于“Elon Musk”的背景信息。
- 路径1：Elon Musk -> 创始人 -> SpaceX -> 首席执行官 -> Elon Musk (循环)
- 路径2：Elon Musk -> 创始人 -> Tesla -> 首席执行官 -> Elon Musk (循环)
- 路径3：Elon Musk -> 投资人 -> Neuralink -> 创始人 -> Elon Musk (循环)
如果我们简单地沿着所有路径收集信息，LLM的上下文窗口可能会收到多次关于“Elon Musk是SpaceX的CEO”、“Elon Musk是Tesla的CEO”等信息，甚至重复的“Elon Musk”实体描述。这显然是低效的。
三、上下文去重的核心原则
为了有效解决上述问题，我们需要遵循以下核心原则：
- 识别冗余： 这不仅仅是字面上的重复，更包括语义上的重复。
- 量化上下文相关性： 并非所有重复都是等价的。某些重复在特定上下文中可能是必要的（例如，为了强调或作为不同路径的证据），而另一些则完全是噪音。我们需要一种机制来评估信息的“价值”或“新鲜度”。
- 动态适应性： 去重策略不应是静态的。它应该能够根据当前的查询、任务目标、已有的上下文内容以及上下文窗口的剩余容量动态调整。
- 保留关键信息： 去重不能导致关键信息的丢失或误解。这是最重要的一点。
接下来，我们将详细探讨实现这些原则的具体技术和编程实践。
四、上下文去重技术：代码与实践
我们将从基础的图遍历去重开始，逐步深入到更复杂的语义和结构化去重技术。以下代码示例将主要使用Python语言，因为它在图处理和文本处理领域有广泛的库支持。
A. 基础图遍历与状态管理
这是最基本的去重机制，主要目的是防止无限循环和重复处理相同的节点/边，但它并非直接针对“上下文窗口信息冗余”。
1. 访问集合 (Visited Sets) – DFS/BFS
在深度优先搜索（DFS）或广度优先搜索（BFS）中，我们通常使用一个 visited 集合来记录已经访问过的节点，以避免重复访问和陷入循环。
from collections import deque
class Graph:
def __init__(self):
self.adj = {} # 邻接表
def add_edge(self, u, v):
if u not in self.adj:
self.adj[u] = []
if v not in self.adj:
self.adj[v] = []
self.adj[u].append(v)
# 如果是无向图，还需要 self.adj[v].append(u)
def get_neighbors(self, node):
return self.adj.get(node, [])
def dfs_traversal(self, start_node):
visited = set()
path = []
def dfs_recursive(node):
if node in visited:
return
visited.add(node)
path.append(node) # 收集节点作为路径的一部分
print(f"Visiting node: {node}") # 模拟信息进入上下文
for neighbor in self.get_neighbors(node):
dfs_recursive(neighbor)
dfs_recursive(start_node)
return path
def bfs_traversal(self, start_node):
visited = set()
queue = deque([start_node])
path = []
visited.add(start_node)
while queue:
node = queue.popleft()
path.append(node) # 收集节点作为路径的一部分
print(f"Visiting node: {node}") # 模拟信息进入上下文
for neighbor in self.get_neighbors(node):
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
return path
# 示例图
graph = Graph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "D")
graph.add_edge("C", "D")
graph.add_edge("D", "E")
graph.add_edge("E", "A") # 引入循环 A->B->D->E->A
print("--- DFS Traversal ---")
dfs_path = graph.dfs_traversal("A")
print(f"DFS Path: {dfs_path}n")
print("--- BFS Traversal ---")
bfs_path = graph.bfs_traversal("A")
print(f"BFS Path: {bfs_path}n")局限性：
visited 集合有效地防止了无限循环，但它只解决了“节点是否已被访问”的问题。对于上下文窗口而言，即使一个节点只被访问一次，其携带的背景信息（如详细描述）如果通过多条不同路径被多次生成并添加到上下文中，visited 集合也无法阻止这种信息层面的冗余。例如，如果节点D有一个长篇描述，无论从A->B->D还是A->C->D到达D，这个描述都会被完整地提取一次。
B. 哈希与指纹：内容层面的去重
为了解决信息层面的冗余，我们可以利用哈希技术。通过对信息块（如节点描述、边属性组合）计算哈希值，可以快速判断两个信息块是否完全相同。
1. 内容哈希
对于文本内容，可以使用MD5、SHA256等哈希算法。
import hashlib
class ContextualDeduplicator:
def __init__(self):
self.seen_hashes = set() # 存储已添加到上下文的信息的哈希值
self.context_items = [] # 模拟最终的上下文窗口内容
self.node_descriptions = {
"A": "Node A is the starting point of our journey, representing a user.",
"B": "Node B is an intermediate step, perhaps a product category.",
"C": "Node C is another intermediate step, maybe a different product category.",
"D": "Node D is a critical entity, a specific product detail page.",
"E": "Node E signifies a purchase completion or a feedback loop. It connects back to A."
}
def add_to_context(self, item_id, content):
"""
尝试将内容添加到上下文。如果内容的哈希值已存在，则跳过。
"""
content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()
if content_hash in self.seen_hashes:
print(f"Skipping duplicate content for {item_id}: '{content[:30]}...' (hash: {content_hash[:8]}...)")
return False
self.seen_hashes.add(content_hash)
self.context_items.append(f"[{item_id}]: {content}")
print(f"Added to context for {item_id}: '{content[:30]}...'")
return True
def get_final_context(self):
return "n".join(self.context_items)
# 模拟图遍历并提取信息
graph = Graph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "D")
graph.add_edge("C", "D")
graph.add_edge("D", "E")
graph.add_edge("E", "A")
deduplicator = ContextualDeduplicator()
# 模拟一个会重复访问D的遍历路径
# Path 1: A -> B -> D
print("n--- Simulating Path 1: A -> B -> D ---")
deduplicator.add_to_context("A", deduplicator.node_descriptions["A"])
deduplicator.add_to_context("B", deduplicator.node_descriptions["B"])
deduplicator.add_to_context("D", deduplicator.node_descriptions["D"])
# Path 2: A -> C -> D
print("n--- Simulating Path 2: A -> C -> D ---")
# A 已经被添加过，B和C不同
# deduplicator.add_to_context("A", deduplicator.node_descriptions["A"]) # 会被跳过
deduplicator.add_to_context("C", deduplicator.node_descriptions["C"])
deduplicator.add_to_context("D", deduplicator.node_descriptions["D"]) # D的描述与之前完全相同，会被跳过
print("n--- Final Context ---")
print(deduplicator.get_final_context())这个方法能有效去除完全相同的内容。
表1：常见哈希算法比较
| 算法 | 长度 (比特) | 冲突概率 | 计算速度 | 主要用途 |
|---|---|---|---|---|
| MD5 | 128 | 中 | 快 | 文件完整性校验，早期数据去重 |
| SHA-1 | 160 | 较高 | 中 | 已不推荐用于安全，但仍可用于非安全去重 |
| SHA-256 | 256 | 低 | 慢 | 数据完整性，数字签名，加密货币 |
| SHA-512 | 512 | 极低 | 较慢 | 高安全性场景 |
| FarmHash | 64/128 | 低 | 极快 | Google出品，非加密哈希，字符串快速哈希 |
| MurmurHash | 32/64/128 | 低 | 极快 | 速度快，分布均匀，非加密哈希 |
局限性：
哈希去重是字面意义上的去重。对于语义相似但字面不同的信息，它无能为力。例如，“Elon Musk是SpaceX的CEO”和“SpaceX的首席执行官是Elon Musk”会被视为两个不同的信息，即使它们表达的是相同的事实。
C. 图感知去重策略
为了应对语义和结构层面的冗余，我们需要更智能的图感知策略。
1. 路径优先剪枝与信息增量
当一个节点或子图可以通过多条路径访问时，我们可以选择一条“最佳”路径来提取其核心信息，而在后续访问中，只提取该节点/子图的新增、补充信息，或仅以引用方式处理。
策略：
- 首次完整提取： 第一次访问某个节点时，将其所有相关背景信息（描述、关键属性等）完整地添加到上下文。
- 后续增量提取/引用： 再次访问该节点时，
- 如果已提供的信息足够，则不再重复添加，或者仅添加指向该节点ID的引用（例如，“见节点D”）。
- 如果存在特定于当前路径的额外信息（例如，通过这条路径发现D有新的角色），则只添加这些增量信息。
- 可以优先考虑最短路径、最近路径或信息量最大的路径。
class GraphContextManager:
def __init__(self, node_descriptions):
self.node_descriptions = node_descriptions
self.context_items = []
self.seen_node_ids = set() # 记录已完整介绍过的节点ID
self.node_id_to_context_idx = {} # 记录节点ID首次出现在上下文中的索引
def add_node_info(self, node_id, current_path_info=""):
"""
将节点信息添加到上下文。
如果节点首次出现，则完整添加其描述。
如果节点已出现，则根据情况添加增量信息或引用。
"""
node_full_desc = self.node_descriptions.get(node_id, f"Unknown node {node_id}.")
if node_id not in self.seen_node_ids:
# 首次完整添加
item = f"Node {node_id}: {node_full_desc}"
if current_path_info:
item += f" (Relevant in context of: {current_path_info})"
self.context_items.append(item)
self.seen_node_ids.add(node_id)
self.node_id_to_context_idx[node_id] = len(self.context_items) - 1
print(f"Added NEW info for {node_id}: '{item[:50]}...'")
return True
else:
# 节点已存在，尝试添加增量信息或生成引用
# 这里的逻辑可以更复杂，例如检查 current_path_info 是否提供了新视角
# 简单起见，这里只生成引用
existing_idx = self.node_id_to_context_idx[node_id]
ref_item = f"Node {node_id} (already introduced, see context item #{existing_idx+1})"
if current_path_info and "already introduced" not in self.context_items[-1]: # 避免连续重复引用
ref_item += f" [Additional context from current path: {current_path_info}]"
# 检查是否上一个添加的也是对这个节点的引用，避免连续的冗余引用
if not self.context_items or not self.context_items[-1].startswith(f"Node {node_id} (already introduced"):
self.context_items.append(ref_item)
print(f"Added REFERENCE for {node_id}: '{ref_item}'")
else:
print(f"Skipping redundant reference for {node_id}.")
return False
def get_final_context(self):
return "n".join(self.context_items)
# 示例图与节点描述
graph_nodes_desc = {
"A": "User A, interested in tech gadgets.",
"B": "Product Category: Smartphones. Known for high-end devices.",
"C": "Product Category: Laptops. Known for productivity machines.",
"D": "Specific Product: Flagship Phone X. Features: AI camera, long battery life, premium design. (Critical information)",
"E": "Review Section for Flagship Phone X. Contains user reviews and ratings.",
"F": "Related Products: Accessories for Phone X (e.g., cases, chargers)."
}
graph = Graph()
graph.add_edge("A", "B")
graph.add_edge("B", "D")
graph.add_edge("A", "C")
graph.add_edge("C", "D") # D可以通过B和C两条路径到达
graph.add_edge("D", "E")
graph.add_edge("E", "F")
context_manager = GraphContextManager(graph_nodes_desc)
print("--- Path 1: A -> B -> D -> E ---")
context_manager.add_node_info("A", "Initial user query")
context_manager.add_node_info("B", "User navigated to Smartphones")
context_manager.add_node_info("D", "User clicked on Flagship Phone X from Smartphones")
context_manager.add_node_info("E", "User viewing reviews")
print("n--- Path 2: A -> C -> D ---")
context_manager.add_node_info("A", "User changed mind to Laptops (implicitly already processed)")
context_manager.add_node_info("C", "User navigated to Laptops")
context_manager.add_node_info("D", "User clicked on Flagship Phone X from Laptops recommendation") # D再次出现
print("n--- Path 3: D -> F (from a different entry point or previous interaction) ---")
context_manager.add_node_info("D", "User now interested in accessories for Phone X") # D再次出现
context_manager.add_node_info("F", "User viewing accessories")
print("n--- Final Context ---")
print(context_manager.get_final_context())这个策略能够避免在上下文窗口中重复填充完整节点描述，而是通过引用或增量信息来处理。
2. 子图规范化与抽象
在某些场景下，图中可能存在频繁出现的、结构相同的子图（Graph Motif）。识别这些子图并对其进行规范化处理，可以进一步减少上下文中的冗余。例如，一个“用户购买流程”的子图可能在不同用户的行为路径中重复出现。
策略：
- 识别频繁子图： 使用图模式挖掘算法（如gSpan, SUBDUE）来发现图中频繁出现的子图结构。
- 抽象表示： 一旦一个频繁子图被识别并完整描述过一次，后续出现时可以将其替换为一个抽象的代表（例如，“标准化购买流程图”）。
- 参数化： 如果子图结构相同但内部节点或边属性不同，可以将其表示为参数化的模板。
这部分通常涉及更复杂的图模式匹配和图数据库操作，代码实现会非常庞大。这里提供一个概念性的Python伪代码，展示如何替换一个已识别的模式。
# 假设我们已经识别出并存储了常见的子图模式
# 例如，一个“产品购买路径”模式： (User) --clicks-> (ProductPage) --adds_to_cart-> (Cart) --checks_out-> (Order)
class SubgraphAbstractor:
def __init__(self):
self.known_subgraph_patterns = {
"product_purchase_flow": {
"nodes": ["User", "ProductPage", "Cart", "Order"],
"edges": [("User", "clicks", "ProductPage"), ("ProductPage", "adds_to_cart", "Cart"), ("Cart", "checks_out", "Order")],
"description": "Standard product purchase flow: User clicks product, adds to cart, checks out."
}
}
self.context_items = []
self.seen_subgraph_hashes = set() # 存储已添加的抽象子图的哈希
def _hash_subgraph_instance(self, instance_nodes, instance_edges):
# 实际实现中，需要一个稳定的方式来哈希一个子图实例，
# 比如先规范化节点顺序和边，再进行哈希。
# 这里简化为对节点和边字符串表示的哈希
normalized_repr = f"nodes:{sorted(instance_nodes)};edges:{sorted([tuple(sorted(e)) for e in instance_edges])}"
return hashlib.sha256(normalized_repr.encode('utf-8')).hexdigest()
def add_subgraph_to_context(self, subgraph_name, instance_nodes, instance_edges):
if subgraph_name not in self.known_subgraph_patterns:
# 如果不是已知模式，则作为普通信息添加
self.context_items.append(f"Unrecognized subgraph: nodes={instance_nodes}, edges={instance_edges}")
print(f"Added unrecognized subgraph.")
return
pattern_info = self.known_subgraph_patterns[subgraph_name]
instance_hash = self._hash_subgraph_instance(instance_nodes, instance_edges)
if instance_hash in self.seen_subgraph_hashes:
print(f"Skipping duplicate instance of '{subgraph_name}' (hash: {instance_hash[:8]}...)")
return False
# 首次出现，添加抽象描述
self.context_items.append(f"Observed pattern '{subgraph_name}': {pattern_info['description']} (Involving: {', '.join(instance_nodes)})")
self.seen_subgraph_hashes.add(instance_hash)
print(f"Added abstract description for '{subgraph_name}' (hash: {instance_hash[:8]}...)")
return True
def get_final_context(self):
return "n".join(self.context_items)
# 模拟
subgraph_abs = SubgraphAbstractor()
# 第一次看到购买流程
print("n--- First Purchase Flow ---")
subgraph_abs.add_subgraph_to_context(
"product_purchase_flow",
instance_nodes=["Alice", "iPhone Page", "Alice's Cart", "Order #123"],
instance_edges=[("Alice", "clicks", "iPhone Page"), ("iPhone Page", "adds_to_cart", "Alice's Cart"), ("Alice's Cart", "checks_out", "Order #123")]
)
# 第二次看到购买流程，节点不同但结构相同
print("n--- Second Purchase Flow (different entities) ---")
subgraph_abs.add_subgraph_to_context(
"product_purchase_flow",
instance_nodes=["Bob", "MacBook Page", "Bob's Cart", "Order #456"],
instance_edges=[("Bob", "clicks", "MacBook Page"), ("MacBook Page", "adds_to_cart", "Bob's Cart"), ("Bob's Cart", "checks_out", "Order #456")]
)
# 假设由于某种原因，又捕捉到与第一次完全相同的实例（例如重放）
print("n--- Third Purchase Flow (identical instance) ---")
subgraph_abs.add_subgraph_to_context(
"product_purchase_flow",
instance_nodes=["Alice", "iPhone Page", "Alice's Cart", "Order #123"],
instance_edges=[("Alice", "clicks", "iPhone Page"), ("iPhone Page", "adds_to_cart", "Alice's Cart"), ("Alice's Cart", "checks_out", "Order #123")]
)
print("n--- Final Context ---")
print(subgraph_abs.get_final_context())这种方法将复杂子图的细节用一个简洁的抽象描述替代，但需要强大的图模式挖掘和匹配能力。
3. 基于语义相似度的过滤
哈希去重无法处理语义相似但字面不同的内容。引入自然语言处理（NLP）技术，特别是词向量或句子嵌入，可以检测语义上的冗余。
策略：
- 生成嵌入： 使用预训练的语言模型（如Sentence-BERT, OpenAI embeddings）将每个背景信息块转化为高维向量（嵌入）。
- 存储与查询： 将这些嵌入存储在向量数据库（如Faiss, Pinecone, Milvus, Weaviate）中。
- 相似度比较： 当有新的信息块要添加到上下文时，计算其与已存在嵌入的相似度（通常是余弦相似度）。
- 阈值判断： 如果相似度超过某个预设阈值，则认为它是冗余的，并进行去重。
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
class SemanticDeduplicator:
def __init__(self, model_name='all-MiniLM-L6-v2', similarity_threshold=0.9):
self.model = SentenceTransformer(model_name)
self.similarity_threshold = similarity_threshold
self.context_items = []
self.context_embeddings = [] # 存储已添加到上下文的项目的嵌入
self.context_texts = [] # 存储已添加到上下文的项目的原始文本
def add_to_context(self, content, item_id=""):
"""
根据语义相似度将内容添加到上下文。
"""
content_embedding = self.model.encode([content])[0] # 编码为向量
if not self.context_embeddings:
# 上下文为空，直接添加
self.context_items.append(f"[{item_id}] {content}")
self.context_embeddings.append(content_embedding)
self.context_texts.append(content)
print(f"Added NEW semantic item for {item_id}: '{content[:50]}...'")
return True
# 计算与已有内容的相似度
similarities = cosine_similarity([content_embedding], self.context_embeddings)[0]
# 检查是否存在高于阈值的相似度
for i, sim in enumerate(similarities):
if sim > self.similarity_threshold:
print(f"Skipping SEMANTICALLY DUPLICATE content for {item_id}: '{content[:30]}...' (similar to '{self.context_texts[i][:30]}...' sim={sim:.2f})")
return False
# 没有发现相似内容，添加
self.context_items.append(f"[{item_id}] {content}")
self.context_embeddings.append(content_embedding)
self.context_texts.append(content)
print(f"Added NEW semantic item for {item_id}: '{content[:50]}...'")
return True
def get_final_context(self):
return "n".join(self.context_items)
# 示例
sem_dedup = SemanticDeduplicator()
print("n--- Semantic Deduplication Test ---")
sem_dedup.add_to_context("Elon Musk is the CEO of SpaceX.", "Fact1")
sem_dedup.add_to_context("The chief executive officer of SpaceX is Elon Musk.", "Fact2") # 语义相似
sem_dedup.add_to_context("Elon Musk also leads Tesla Motors.", "Fact3") # 新信息
sem_dedup.add_to_context("Tesla Motors is spearheaded by Elon Musk.", "Fact4") # 语义相似
sem_dedup.add_to_context("SpaceX is a private American aerospace manufacturer and space transportation services company.", "Fact5") # 新信息
sem_dedup.add_to_context("Musk's company, Neuralink, focuses on brain-computer interfaces.", "Fact6") # 新信息
print("n--- Final Semantic Context ---")
print(sem_dedup.get_final_context())局限性：
- 计算成本： 生成嵌入和计算相似度是耗时的，尤其是在处理大量背景信息时。
- 阈值选择： 相似度阈值的选择至关重要，过高可能导致漏掉冗余，过低可能删除重要但略有不同的信息。
- 模型局限： 嵌入模型可能无法捕捉所有细微的语义差异，或在特定领域表现不佳。
4. 上下文相关性评分
结合上述技术，我们可以为每个潜在的上下文信息项分配一个“相关性分数”，并根据这个分数和上下文窗口的容量进行动态选择。
策略：
- 多维度评分： 结合多个因素：
- 距离： 从当前查询节点到信息源节点的图距离。
- 路径重要性： 信息所在路径的类型（例如，直接关系比间接关系更重要）。
- 信息类型： 核心属性比辅助属性更重要。
- 新颖性： 信息是否已在上下文中出现过（哈希或语义去重得分）。
- 时间戳： 对于动态图，最近的信息可能更重要。
- 用户偏好/任务目标： 针对特定查询或任务，某些类型的信息权重更高。
- 动态选择： 根据上下文窗口的剩余容量，选择得分最高且未被严格去重的信息。
class ContextualRelevanceScorer:
def __init__(self, max_context_tokens=1000, token_estimator=lambda x: len(x.split())):
self.max_context_tokens = max_context_tokens
self.token_estimator = token_estimator # 简单的单词计数器
self.current_context_tokens = 0
self.final_context_items = []
self.candidate_items = [] # 存储所有候选信息及其分数
self.seen_content_hashes = set() # 用于内容去重
self.node_embeddings = {} # 模拟节点嵌入，用于计算距离/相似度
def add_candidate_info(self, item_id, content, path_distance, item_type="general", is_new_info=True):
"""
添加一个候选信息项，并计算其初始相关性分数。
- item_id: 信息的唯一标识
- content: 信息的文本内容
- path_distance: 从查询点到该信息源的图距离 (越近越重要)
- item_type: 信息的类型 (例如 "core_fact", "detail", "relation")
- is_new_info: 是否是全新的信息 (基于哈希或语义去重)
"""
content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()
if content_hash in self.seen_content_hashes:
print(f"Candidate {item_id} is a hard duplicate, skipping initial consideration.")
return
score = 0
# 1. 距离分数 (距离越近，分数越高)
score += max(0, 10 - path_distance)
# 2. 类型分数
if item_type == "core_fact":
score += 5
elif item_type == "detail":
score += 2
elif item_type == "relation":
score += 3
# 3. 新颖性分数 (如果信息是全新的，加分)
if is_new_info:
score += 7
# 4. (可选) 语义相关性分数：与当前查询的相似度，这里简化为常数
score += 2 # 假设与查询有一定相关性
self.candidate_items.append({
"id": item_id,
"content": content,
"score": score,
"tokens": self.token_estimator(content),
"hash": content_hash
})
self.seen_content_hashes.add(content_hash)
print(f"Added candidate {item_id} with score {score}: '{content[:40]}...'")
def select_context_items(self):
"""
根据分数和令牌限制，选择最终的上下文项。
"""
# 按照分数降序排列
self.candidate_items.sort(key=lambda x: x["score"], reverse=True)
selected_hashes = set()
for item in self.candidate_items:
# 再次检查硬哈希去重，防止低分重复项被意外选中
if item["hash"] in selected_hashes:
continue
if self.current_context_tokens + item["tokens"] <= self.max_context_tokens:
self.final_context_items.append(f"[{item['id']}] (Score: {item['score']}) {item['content']}")
self.current_context_tokens += item["tokens"]
selected_hashes.add(item["hash"]) # 标记为已选择
else:
print(f"Context window full. Skipping {item['id']}.")
break # 上下文窗口已满，停止添加
def get_final_context(self):
return "n".join(self.final_context_items)
# 示例
scorer = ContextualRelevanceScorer(max_context_tokens=100)
print("n--- Adding Candidate Information ---")
scorer.add_candidate_info("NodeA_desc", "Node A is the root of our graph, a user profile.", path_distance=0, item_type="core_fact", is_new_info=True)
scorer.add_candidate_info("NodeB_desc", "Node B describes a product category: Electronics.", path_distance=1, item_type="detail", is_new_info=True)
scorer.add_candidate_info("NodeC_desc", "Node C describes another category: Books.", path_distance=1, item_type="detail", is_new_info=True)
scorer.add_candidate_info("NodeD_desc_path1", "Node D is a specific item: 'The Art of Programming'.", path_distance=2, item_type="core_fact", is_new_info=True)
scorer.add_candidate_info("NodeD_desc_path2", "Node D is a specific item: 'The Art of Programming'.", path_distance=2, item_type="core_fact", is_new_info=False) # 假设这是通过另一条路径发现的，内容相同
scorer.add_candidate_info("EdgeAB_rel", "Relationship: A is interested in B.", path_distance=1, item_type="relation", is_new_info=True)
scorer.add_candidate_info("EdgeAC_rel", "Relationship: A also browses C.", path_distance=1, item_type="relation", is_new_info=True)
scorer.add_candidate_info("NodeX_low_relevance", "This is a very distant and less relevant node about historical data.", path_distance=10, item_type="detail", is_new_info=True)
print("n--- Selecting Context Items ---")
scorer.select_context_items()
print("n--- Final Scored Context ---")
print(scorer.get_final_context())
print(f"Total tokens used: {scorer.current_context_tokens}/{scorer.max_context_tokens}")相关性评分是高度灵活且强大的，它允许我们根据多种因素进行精细控制，以在有限的上下文窗口中最大化信息价值。
5. 增量上下文构建与引用
这是将上述思想应用于LLM上下文的一种常见模式。一旦一个实体被完整描述，后续的提及可以将其替换为简短的ID或别名。
策略：
- 实体识别与链接： 识别文本中的命名实体，并将其链接到图中的节点。
- 上下文映射： 维护一个当前上下文窗口内已提及实体到其ID或简称的映射。
- 替换： 在生成后续信息时，如果提及已在映射中的实体，则用其简称或ID替换完整描述。
class IncrementalContextBuilder:
def __init__(self, max_tokens=200, token_estimator=lambda x: len(x.split())):
self.max_tokens = max_tokens
self.token_estimator = token_estimator
self.current_tokens = 0
self.context_lines = []
self.entity_map = {} # 存储已引入的实体ID及其简介
self.entity_counter = 1 # 用于生成唯一的实体引用ID
def add_entity_description(self, entity_id, full_description):
"""
添加一个实体的完整描述。
如果实体已引入，则跳过或仅更新。
"""
if entity_id in self.entity_map:
print(f"Entity '{entity_id}' already introduced as '{self.entity_map[entity_id]['alias']}'. Skipping full description.")
return
# 检查是否还有足够空间添加完整描述
desc_tokens = self.token_estimator(full_description)
if self.current_tokens + desc_tokens > self.max_tokens:
print(f"Not enough space for full description of '{entity_id}'. Max tokens reached.")
return
alias = f"E{self.entity_counter}"
self.entity_map[entity_id] = {"alias": alias, "description": full_description}
self.entity_counter += 1
line = f"Entity {alias} ({entity_id}): {full_description}"
self.context_lines.append(line)
self.current_tokens += self.token_estimator(line)
print(f"Introduced entity '{entity_id}' as '{alias}'. Tokens: {self.current_tokens}/{self.max_tokens}")
return alias
def add_relationship(self, entity1_id, relation, entity2_id, details=""):
"""
添加实体间的关系，使用别名进行引用。
"""
alias1 = self.entity_map.get(entity1_id, {}).get("alias", entity1_id)
alias2 = self.entity_map.get(entity2_id, {}).get("alias", entity2_id)
rel_line = f"Relationship: {alias1} {relation} {alias2}. {details}".strip()
rel_tokens = self.token_estimator(rel_line)
if self.current_tokens + rel_tokens > self.max_tokens:
print(f"Not enough space for relationship: {rel_line[:50]}... Max tokens reached.")
return
self.context_lines.append(rel_line)
self.current_tokens += rel_tokens
print(f"Added relationship: '{rel_line}'. Tokens: {self.current_tokens}/{self.max_tokens}")
def get_final_context(self):
return "n".join(self.context_lines)
# 示例
builder = IncrementalContextBuilder(max_tokens=100)
print("n--- Building Context Incrementally ---")
builder.add_entity_description("ElonMusk", "Elon Musk is a business magnate, investor, and engineer. He is the founder, CEO, and chief engineer of SpaceX.")
builder.add_entity_description("SpaceX", "SpaceX is an American spacecraft manufacturer, launch service provider, and satellite communications company.")
builder.add_entity_description("Tesla", "Tesla, Inc. is an American multinational automotive and clean energy company headquartered in Austin, Texas.")
builder.add_relationship("ElonMusk", "is CEO of", "SpaceX", "This is his primary role there.")
builder.add_relationship("ElonMusk", "is CEO of", "Tesla", "He also leads the electric vehicle company.")
builder.add_entity_description("SpaceX", "SpaceX also develops Starlink.", ) # 再次尝试添加 SpaceX 描述
builder.add_relationship("SpaceX", "develops", "Starlink", "Starlink is a satellite internet constellation.") # 引入 Starlink
builder.add_entity_description("Starlink", "Starlink is a satellite internet constellation operated by SpaceX, providing satellite Internet access coverage to over 60 countries.") # Starlink 完整描述
print("n--- Final Incremental Context ---")
print(builder.get_final_context())
print(f"Total tokens used: {builder.current_tokens}/{builder.max_tokens}")这种方法对于生成LLM提示非常有效，因为它可以在有限的令牌预算内最大化地传递信息，并通过清晰的引用机制维持实体间的联系。
6. 基于时间/新鲜度的去重（适用于动态图）
在动态图或实时数据流中，信息的“新鲜度”可以作为去重的重要标准。即使是重复的信息，如果其时间戳非常新，可能意味着它具有新的相关性或更新的状态。
策略：
- 时间戳： 为每个信息项关联一个时间戳。
- 滑动窗口： 仅考虑在特定时间窗口内的信息进行去重。
- 优先最新： 如果存在多个重复项，优先保留最新鲜的。
import time
class TemporalDeduplicator:
def __init__(self, time_window_seconds=3600): # 1小时内重复的信息被视为冗余
self.time_window_seconds = time_window_seconds
self.context_items = []
self.seen_content_with_timestamp = {} # {content_hash: latest_timestamp}
def add_to_context(self, content, item_id=""):
content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest()
current_time = time.time()
if content_hash in self.seen_content_with_timestamp:
last_seen_time = self.seen_content_with_timestamp[content_hash]
if (current_time - last_seen_time) < self.time_window_seconds:
print(f"Skipping TEMPORAL DUPLICATE content for {item_id}: '{content[:30]}...' (seen {current_time - last_seen_time:.0f}s ago)")
# 更新时间戳，表示此信息仍然活跃
self.seen_content_with_timestamp[content_hash] = current_time
return False
else:
# 超过时间窗口，可以重新添加或视为新信息
print(f"Content for {item_id} is old, re-adding: '{content[:30]}...' (last seen {current_time - last_seen_time:.0f}s ago)")
self.context_items.append(f"[{item_id} @ {time.ctime(current_time)}] {content}")
self.seen_content_with_timestamp[content_hash] = current_time
return True
else:
self.context_items.append(f"[{item_id} @ {time.ctime(current_time)}] {content}")
self.seen_content_with_timestamp[content_hash] = current_time
print(f"Added NEW temporal item for {item_id}: '{content[:30]}...'")
return True
def get_final_context(self):
return "n".join(self.context_items)
# 示例
temp_dedup = TemporalDeduplicator(time_window_seconds=10) # 模拟10秒的短时间窗口
print("n--- Temporal Deduplication Test ---")
temp_dedup.add_to_context("User Alice logged in.", "Event1")
time.sleep(2)
temp_dedup.add_to_context("User Alice performed action X.", "Event2")
time.sleep(3)
temp_dedup.add_to_context("User Alice logged in.", "Event3") # 在10秒内重复
time.sleep(12) # 超过10秒窗口
temp_dedup.add_to_context("User Alice logged in.", "Event4") # 再次添加
print("n--- Final Temporal Context ---")
print(temp_dedup.get_final_context())这种方法特别适用于监控、日志分析或实时推荐系统等场景。
V. 架构集成考量
将上下文去重技术集成到实际系统中，需要考虑以下架构层面：
-
预处理 vs. 实时处理：
- 预处理： 对于相对静态的图数据，可以在数据加载或更新时进行离线去重和规范化，将结果存储为更紧凑的形式（如知识图谱中的规范化三元组）。这减少了实时处理的负担。
- 实时处理： 对于动态生成或需要高度上下文敏感的信息，去重逻辑需要在图遍历或信息提取阶段实时执行。语义去重和相关性评分通常属于此范畴。
-
存储机制：
- 图数据库（Neo4j, ArangoDB等）： 存储原始图结构，去重逻辑在查询时应用。
- 向量数据库（Faiss, Pinecone等）： 存储信息块的嵌入，用于语义相似度查询。
- 键值存储（Redis, Memcached）： 存储哈希值或已引入实体ID，用于快速查找。
- 知识图谱： 规范化的知识图谱本身就具有一定的去重和抽象能力。
-
API设计：
- 信息提供者API： 提供一个统一的接口，供图遍历器、LLM代理或其他组件调用，以“请求”背景信息。该API内部封装去重逻辑。
- 上下文管理器API： 暴露
add_to_context,get_final_context等方法，允许客户端逐步构建上下文，并由管理器处理去重。 - 可配置性： API应允许配置去重策略（例如，相似度阈值、相关性权重等）。
-
反馈循环：
- LLM反馈： LLM在推理后，如果能提供哪些背景信息是多余的或缺失的反馈，可以用于优化去重策略。
- 用户反馈： 用户对系统输出的满意度也可以作为去重效果的衡量标准。
VI. 挑战与局限性
尽管上下文去重是至关重要的，但它也面临一些挑战：
- “冗余”的定义模糊性： 什么是真正的冗余？在某些情况下，重复的提及可能用于强调、提供证据或从不同角度确认事实。过度去重可能导致关键信息的丢失或上下文断裂。
- 计算成本： 尤其是语义去重和相关性评分，涉及到复杂的NLP模型和向量运算，可能消耗大量计算资源。
- 损失细微差别： 严格的去重可能抹去信息之间的细微差异，而这些差异在特定上下文中可能很重要。例如，两个实体描述可能语义相似，但其中一个包含时间戳，另一个没有。
- 可伸缩性： 对于拥有数十亿节点和边的超大规模图，管理所有已访问信息的状态、计算哈希或嵌入的成本会非常高昂。
- 领域特定性： 最优的去重策略可能因领域而异。例如，法律文本与社交媒体数据需要不同的相似度阈值和相关性权重。
VII. 未来方向
- LLM驱动的去重： 利用LLM本身的理解能力，让它们直接判断给定上下文中的信息哪些是冗余的、哪些是关键的。这可能通过少样本学习或指令微调实现。
- 强化学习： 训练智能体通过与环境（如LLM推理、用户反馈）的互动，学习最佳的上下文去重策略，以最大化信息密度和任务成功率。
- 可解释性AI： 开发工具来解释为什么某个信息被认为是冗余并被删除，或者为什么某个信息被保留下来，以增加系统的透明度和可信度。
- 混合模型： 结合符号推理（基于图结构）和神经模型（基于语义）的优势，构建更强大、更灵活的去重系统。
结语
在处理大规模循环图并将其信息高效地输入到上下文窗口时，Contextual De-duplication 不仅是性能优化的关键，更是实现智能系统准确、高效理解和响应的基石。通过结合图遍历优化、哈希指纹、语义相似度分析、上下文相关性评分以及增量构建策略，我们能够有效管理信息密度，确保有限的上下文资源被用于传递最有价值、最独特的内容。未来的研究将进一步探索如何利用AI自身的智能，更精细化地定义和执行这一关键任务。