用空间节点画布解决LLM上下文漂移问题

1. 背景与核心概念

1.1 LLM 上下文漂移到底是什么

先从一个很常见的开发场景说起。

很多团队在接入大模型 LLM 后，最初跑通的都是单轮问答：用户提问，模型回答，看起来效果不错。但一旦进入真实业务，比如做一个 AI 编程助手、智能客服、或者知识库问答系统，就很容易发现一个诡异的问题——随着对话轮数增加，模型开始“忘事”。

前两轮你问了“帮我分析订单表结构”，模型回答得很好；第五轮你问“基于刚才说的订单表，写一个按月统计销售额的 SQL”，模型反而开始答非所问，甚至重新定义了一遍订单表。你以为是自己提示词写得不好，其实是**上下文漂移（context drift）**在捣乱。

上下文漂移是指在一次多轮交互或长文档处理过程中，真正关键的信息逐渐被无关内容淹没，导致模型生成质量下降的现象。它并不等同于“上下文窗口不够大”。即使模型支持 128K、200K token，只要对话里混入了大量早期话题、重复解释、或者被噪声信息占据，模型一样会丢失重点。

从 LLM 工作原理来看，Transformer 架构本身对序列中不同位置的注意力权重分布不同，长序列中靠后的内容往往对生成结果影响更大。也就是说，如果你在两个关键事实之间插入了十几轮闲聊，模型很可能把闲聊当作最新状态，把关键事实当作“历史残留”。

1.2 上下文漂移的典型表现

实际项目里，上下文漂移通常有下面几种表现：

这些问题的共同点是：模型不是“不知道”，而是“没找到重点”。传统 RAG 解决的是“模型不知道的外部知识检索”问题，而 context drift 解决的是“模型已经知道但无法聚焦”的问题。

1.3 为什么想到用空间节点画布

面对上下文漂移，常见的解决思路有两种：

1. 滑动窗口：只保留最近 N 轮对话。问题在于，关键信息可能出现在很早的位置，窗口一滑就被丢掉了。
2. 总结压缩：把历史对话压缩成摘要。问题在于，摘要会损失细节，而且摘要本身也会越滚越长。

这两种思路本质上都是用线性方式管理非线性信息。但人的思维方式并不是线性的：你写代码时会想到之前看过的某篇文档，做需求时会关联到某个历史决策，排查 bug 时会跳跃到几天前的一次配置变更。这些关系天然是网状的、空间化的。

空间节点画布（Spatial Node Canvas）的核心思路是：把对话、文档、决策、代码片段全部建模为画布上的节点，用空间位置和连线表达它们之间的关系。模型不再面向一整条流水账去理解上下文，而是面向一个经过筛选和排序的“焦点子图”去理解当前任务。

这个思路很像知识库工具里经常提到的“双链笔记”理念，但用在 LLM 上下文管理上，多了一层“可编程组装”的意义：不仅人能看，程序也能根据当前任务自动选择节点、组装上下文、发送给模型。

2. 系统设计思路

2.1 核心数据模型

空间节点画布要能替代“纯文本历史记录”，首先得定义清楚数据模型。

我建议的最小模型包含以下几类：

• 节点（Node）：上下文中的最小信息单元。可以是一条用户消息、一条模型回复、一段文档摘录、一个代码片段或一条决策记录。
• 边（Edge）：节点之间的关系。比如“引用”“回复”“依赖”“补充”等。
• 画布（Canvas）：节点的空间容器，每个节点有 x、y 坐标和所属画布 ID。
• 焦点节点（Focus Node）：当前正在处理的节点，上下文组装时以它为起点向外扩散。

在实际实现里，节点可以很简单。下面是一个 Python dataclass 定义：

from dataclasses import dataclass, field from typing import Optional, List from uuid import uuid4 import time @dataclass class CanvasNode: """ 画布节点 """ node_id: str = field(default_factory=lambda: str(uuid4())) canvas_id: str = "default" x: float = 0.0 y: float = 0.0 node_type: str = "note" # note / code / doc / chat content: str = "" parent_id: Optional[str] = None tags: List[str] = field(default_factory=list) create_time: float = field(default_factory=time.time) update_time: float = field(default_factory=time.time) @dataclass class CanvasEdge: """ 连接边，表示节点之间的关系 """ source_id: str target_id: str relation: str = "related" # related / reply_to / depends_on / references @dataclass class Canvas: """ 画布本身 """ canvas_id: str = "default" nodes: dict = field(default_factory=dict) # node_id -> CanvasNode edges: list = field(default_factory=list) # list[CanvasEdge]

这里有几个设计细节值得说明：

1. 节点必须有类型和标签。组装上下文时，程序可以根据类型决定是否把它放入系统提示词，例如code节点更适合放在“当前代码上下文”区域，而不是“用户指令”区域。
2. 边必须有关系类型。不同关系决定上下文中的呈现方式，reply_to的节点应该按对话时间排序，depends_on的节点应该放在前置位置。
3. 坐标不是摆设。虽然组装上下文时主要用的是图结构，但坐标可以用于前端渲染、话题聚类、以及用户手动调整布局，是“空间”体验的基础。

2.2 上下文组装流程

空间画布解决 context drift 的关键，在于“组装（assembly）”而不是“拼接（concatenation）”。

一个标准流程如下：

1. 用户在当前节点输入新的消息，程序把这个消息创建为一个新节点。
2. 程序从新节点出发，沿边向外扩展，找到关联节点集合。
3. 按关系类型做排序和分组，形成结构化的上下文。
4. 估算 token 总数，根据模型窗口做截断或降级（去掉低优先级节点）。
5. 把节点内容按“焦点优先、时间顺序、关系分组”的规则写入最终的 prompt。
6. 调用 LLM API，得到回复后把回复也写入画布，挂到当前节点下。

如果用一段伪代码表达，大致是这样：

def assemble_context(canvas: Canvas, focus_node_id: str, max_tokens: int = 8000) -> str: # 1. 从焦点节点扩展获取子图 node_ids = expand_subgraph(canvas, focus_node_id, depth=2) # 2. 将节点按优先级分组 system_nodes = [] history_nodes = [] for nid in node_ids: node = canvas.nodes[nid] if node.node_type == "code": system_nodes.append(node) # 代码上下文放在前面 else: history_nodes.append(node) # 对话历史按时间排序 # 3. 估算 token（简化逻辑，实际场景建议用 tiktoken 或 tokenizer） total_tokens = sum(estimate_tokens(n.content) for n in system_nodes + history_nodes) # 4. 超限裁剪：去掉最外围节点 while total_tokens > max_tokens and history_nodes: popped = history_nodes.pop(0) total_tokens -= estimate_tokens(popped.content) # 5. 拼接文本 sections = [] if system_nodes: sections.append("## 参考节点\n" + "\n".join(n.content for n in system_nodes)) if history_nodes: sections.append("## 对话记录\n" + "\n".join(n.content for n in history_nodes)) return "\n\n".join(sections)

这个流程的核心价值在于：上下文中的内容不是“所有历史”，而是“与当前焦点相关且经过排序的内容”。再加上 token 预算约束，模型中每一次生成所看到的上下文，都是经过剪辑的“特写镜头”，而不是没有重点的“全景录像”。

2.3 技术栈选型

实现这样一个系统，技术栈可以自由选择。如果目标是快速验证，我建议采用“轻前端 + Python 后端”的组合：

• 前端：React + React Flow（或类似节点编辑器库）用来渲染画布、拖拽节点、缩放。
• 后端：Python + FastAPI 提供节点 CRUD、上下文组装、LLM 调用代理。
• 存储：SQLite 起步即可，数据量上来后可换 PostgreSQL。
• 向量检索（可选）：如果要实现“自动寻找相关节点”，可以用 embedding 模型做语义匹配。

这里需要强调一点：所有版本都要根据你的项目实际情况调整。前端框架主版本更新较快，后端依赖也经常变化，本文示例重点讲实现思路，而不是固定某个版本。

3. 环境准备与项目结构

3.1 环境说明

示例代码以 Python 为主，适合先在后端验证核心逻辑。建议环境如下：

• Python 3.10 及以上
• 任意 OpenAI 兼容的 Chat Completions 接口（也可以换成其他 LLM 服务）
• 预留一个用于 embedding 的模型（用于语义相似度计算，不强制）
• 前端部分（可选）：React 18+ / React Flow 的最新稳定版

不需要一开始就把 Node.js 前端搭好。可以先写一个纯 Python 的命令行验证程序，跑通“节点创建 → 上下文组装 → LLM 调用 → 结果写回”的闭环，再补前端。

3.2 项目目录结构

建议按下面的结构组织代码：

spatial-llm-canvas/ ├── app/ │ ├── __init__.py │ ├── models.py # 数据模型 │ ├── canvas_store.py # 画布存储（示例用字典 + JSON 持久化） │ ├── drift_detector.py # 上下文漂移检测 │ ├── context_assembler.py # 上下文组装器 │ └── llm_client.py # LLM 客户端封装 ├── frontend/ # 前端（可选） │ ├── package.json │ └── src/ │ ├── CanvasView.tsx │ └── NodeCard.tsx ├── examples/ │ └── demo.py # 命令行演示脚本 └── requirements.txt

如果你暂时不想写前端，直接把精力放在app/目录和examples/demo.py上就好。

3.3 requirements.txt

fastapi>=0.104.0 uvicorn[standard]>=0.24.0 httpx>=0.25.0 tiktoken>=0.5.0 # 用于 token 估算 numpy>=1.24.0 # 用于相似度计算

如果后端不打算用 FastAPI，也可以把核心逻辑写成一个 Python 脚本直接跑。

4. 完整实战案例

下面进入正题：写一个最小可用的“空间节点画布 + LLM 上下文管理”原型。

为了让演示可复现，我把它拆成几个模块：数据模型、存储层、漂移检测器、上下文组装器、LLM 客户端，最后用一个命令行脚本串联整个流程。

4.1 定义数据模型与存储层

首先建立数据模型，代码放在app/models.py：

# app/models.py import json import time from dataclasses import dataclass, field, asdict from typing import List, Optional, Dict, Any from uuid import uuid4 @dataclass class CanvasNode: node_id: str = field(default_factory=lambda: str(uuid4())) canvas_id: str = "default" x: float = 0.0 y: float = 0.0 node_type: str = "note" content: str = "" parent_id: Optional[str] = None tags: List[str] = field(default_factory=list) create_time: float = field(default_factory=time.time) update_time: float = field(default_factory=time.time) def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class CanvasEdge: source_id: str target_id: str relation: str = "related"

为了简单起见，存储层先用一个类内存实现，再提供 JSON 文件落盘方法。代码放在app/canvas_store.py：

# app/canvas_store.py import json from typing import Dict, List, Optional from .models import CanvasNode, CanvasEdge class CanvasStore: def __init__(self, storage_path: str = "canvas_data.json"): self.storage_path = storage_path self.nodes: Dict[str, CanvasNode] = {} self.edges: List[CanvasEdge] = [] self.canvas_id = "default" self._load() def add_node(self, content: str, node_type: str = "note", x: float = 0.0, y: float = 0.0, parent_id: Optional[str] = None, tags: Optional[List[str]] = None) -> CanvasNode: node = CanvasNode( canvas_id=self.canvas_id, content=content, node_type=node_type, x=x, y=y, parent_id=parent_id, tags=tags or [], ) self.nodes[node.node_id] = node self._save() return node def add_edge(self, source_id: str, target_id: str, relation: str = "related") -> None: self.edges.append(CanvasEdge(source_id=source_id, target_id=target_id, relation=relation)) self._save() def get_neighbors(self, node_id: str) -> List[str]: neighbors = [] for edge in self.edges: if edge.source_id == node_id: neighbors.append(edge.target_id) elif edge.target_id == node_id: neighbors.append(edge.source_id) return neighbors def get_node(self, node_id: str) -> Optional[CanvasNode]: return self.nodes.get(node_id) def _save(self) -> None: with open(self.storage_path, "w", encoding="utf-8") as f: json.dump({ "nodes": [n.to_dict() for n in self.nodes.values()], "edges": [ {"source_id": e.source_id, "target_id": e.target_id, "relation": e.relation} for e in self.edges ], }, f, ensure_ascii=False, indent=2) def _load(self) -> None: try: with open(self.storage_path, "r", encoding="utf-8") as f: data = json.load(f) for item in data.get("nodes", []): node = CanvasNode(**item) self.nodes[node.node_id] = node for item in data.get("edges", []): self.edges.append(CanvasEdge(**item)) except FileNotFoundError: pass

这个存储层的设计有几个好处：

• 方便测试：如果不传storage_path，会默认读写当前目录下的canvas_data.json，你可以随时查看落盘内容，调试非常直观。
• 面向后续扩展：以后要接入 PostgreSQL，只需要把这个类的接口实现换掉，上层逻辑不需要改动。

4.2 上下文漂移检测器

漂移检测的目的，是在“模型开始胡说八道之前”给用户一个提醒。最简单的检测方式是比较新输入与当前焦点主题的语义相似度。

如果手头有 embedding 服务，可以直接用：

# app/drift_detector.py from typing import List, Optional try: import numpy as np except ImportError: np = None # 如果没有 numpy，相似度计算降级为字符串匹配 class DriftDetector: """ 用 embedding 判断新输入是否与当前会话焦点发生漂移。 这里通过一个可插拔的 embed_func 注入 embedding 计算逻辑。 """ def __init__(self, embed_func=None, threshold: float = 0.55): """ :param embed_func: callable, 输入是文本列表，输出是向量列表 :param threshold: 低于该值认为发生主题漂移 """ self.embed_func = embed_func self.threshold = threshold def is_drift(self, new_text: str, reference_texts: List[str]) -> Optional[bool]: if self.embed_func is None: return None # 没有 embedding 能力时不判断，交给后续人工判断 if not reference_texts: return False new_vec = np.array(self.embed_func([new_text])[0]) ref_vecs = np.array(self.embed_func(reference_texts)) # 余弦相似度 scores = [] for ref_vec in ref_vecs: similarity = np.dot(new_vec, ref_vec) / ( np.linalg.norm(new_vec) * np.linalg.norm(ref_vec) + 1e-9 ) scores.append(float(similarity)) max_score = max(scores) return max_score < self.threshold

这里的embed_func可以是任意 embedding API 的封装。使用方式类似：

def my_embed(texts: List[str]) -> List[List[float]]: # 调用你的 embedding 服务 pass detector = DriftDetector(embed_func=my_embed, threshold=0.55) drift = detector.is_drift("现在帮我写订单查询SQL", ["用户说要分析订单表结构", "订单表有订单ID、客户ID、金额"])

需要注意的是：阈值很关键。阈值设得太低，漂移检测不灵敏；设得太高，则会把正常的发散思维误判为主题切换。建议先用测试集统计分布，再确定阈值。

如果暂时没有 embedding 模型，也可以用简单的 Jaccard 文本相似度作为弱化版，但效果会明显下降，适合 demo 阶段使用。

4.3 上下文组装器

上下文组装器是整个系统的核心。它需要做三件事：找关联节点、按优先级排序、执行 token 预算裁剪。

# app/context_assembler.py import heapq from collections import deque from typing import Dict, List from .models import CanvasNode from .canvas_store import CanvasStore class ContextAssembler: def __init__(self, max_tokens: int = 8000, token_estimator=None): self.max_tokens = max_tokens self.token_estimator = token_estimator or self._default_estimate def _default_estimate(self, text: str) -> int: # 中文场景下粗略按 1 个汉字 ≈ 1 token 估算，英文场景可按单词数 × 1.3 if not text: return 0 chinese_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff') english_words = len([w for w in text.split() if any(c.isalpha() for c in w)]) return chinese_chars + int(english_words * 1.3) def collect_nodes(self, store: CanvasStore, focus_node_id: str, depth_limit: int = 3) -> List[str]: """ 从焦点节点出发，BFS 遍历图，返回节点 id 列表。 优先返回直接邻居，再返回间接邻居。 """ visited = set() queue = deque([(focus_node_id, 0)]) result = [] while queue: nid, depth = queue.popleft() if nid in visited or depth > depth_limit: continue visited.add(nid) result.append(nid) for neighbor in store.get_neighbors(nid): if neighbor not in visited: queue.append((neighbor, depth + 1)) return result def assemble(self, store: CanvasStore, focus_node_id: str, with_system_prefix: bool = True) -> str: node_ids = self.collect_nodes(store, focus_node_id) nodes = [store.get_node(nid) for nid in node_ids] nodes = [n for n in nodes if n is not None] # 按节点类型分区 system_nodes = [n for n in nodes if n.node_type == "code"] doc_nodes = [n for n in nodes if n.node_type == "doc"] chat_nodes = [n for n in nodes if n.node_type == "chat"] # 对话节点按创建时间排序 chat_nodes.sort(key=lambda n: n.create_time) # 估算 token 并裁剪 total_tokens = 0 used_nodes = [] for n in system_nodes + doc_nodes + chat_nodes: tokens = self.token_estimator(n.content) if total_tokens + tokens > self.max_tokens: break used_nodes.append(n) total_tokens += tokens # 拼接 sections = [] if with_system_prefix: sections.append("你是一个基于项目上下文进行回答的助手。") if system_nodes: sections.append("【代码参考】\n" + "\n\n".join(n.content for n in system_nodes)) if doc_nodes: sections.append("【文档参考】\n" + "\n\n".join(n.content for n in doc_nodes)) if chat_nodes: sections.append("【对话历史】\n" + "\n\n".join(n.content for n in chat_nodes)) sections.append("【当前问题】\n" + store.get_node(focus_node_id).content) return "\n\n".join(sections)

这里最值得留意的是token_estimator。真实项目中，建议直接用tiktoken或模型自带的 tokenizer，而不是用这种粗略估算。因为 token 超限会导致 API 调用直接报错。示例里的估算函数只是为了让你在没有 tokenizer 的情况下也能跑通流程。

组装器还有一个可以扩展的点：裁剪策略。上面的实现是“按固定顺序放入，放不下就丢弃”。更合理的方式是先按与焦点节点的距离打分，距离近的节点优先保留，然后对同一距离的节点再按重要程度排序。这样即使在 token 紧张时，也能保证和当前问题关系最近的上下文被保留下来。

4.4 LLM 客户端封装

LLM 调用不建议直接散落在业务代码里，可以抽一个简单的封装层。下面以 OpenAI 兼容接口为例：

# app/llm_client.py import httpx class LLMClient: def __init__(self, api_key: str, base_url: str = None, model: str = "gpt-4o-mini"): self.api_key = api_key self.base_url = base_url or "https://api.openai.com/v1" self.model = model def chat(self, system_prompt: str, user_prompt: str, temperature: float = 0.3) -> str: url = f"{self.base_url}/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": temperature, } with httpx.Client(timeout=60) as client: resp = client.post(url, headers=headers, json=payload) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"]

这里有几个工程上的注意点：

• API Key 不要写死在代码里，建议用环境变量或本地配置文件管理。
• 调用前先检查resp.status_code，出现 429 限流时需要退避重试。
• 生产环境建议加一层超时控制和异常处理，避免模型接口抖动导致整个服务失败。

4.5 前端交互画布示例（可选）

如果你要做一个真正可视化的“空间节点画布”，前端必不可少。React Flow 是一个比较成熟的节点编辑库，可以用它快速实现拖拽、连线、缩放。

下面是一个最简的 React 组件示意，不追求完整运行，只展示思路：

// frontend/src/CanvasView.tsx import React, { useCallback } from "react"; import ReactFlow, { addEdge, Background, Connection, Edge, Node, useEdgesState, useNodesState, } from "reactflow"; import "reactflow/dist/style.css"; const initialNodes: Node[] = [ { id: "1", position: { x: 100, y: 100 }, data: { label: "需求文档" } }, { id: "2", position: { x: 350, y: 250 }, data: { label: "订单表结构" } }, { id: "3", position: { x: 600, y: 120 }, data: { label: "SQL 写法" } }, ]; const initialEdges: Edge[] = [ { id: "e1-2", source: "1", target: "2", label: "引用" }, { id: "e2-3", source: "2", target: "3", label: "依赖" }, ]; export function CanvasView() { const [nodes, setNodes, onNodesChange] = useNodesState(initialNodes); const [edges, setEdges, onEdgesChange] = useEdgesState(initialEdges); const onConnect = useCallback( (connection: Connection) => setEdges((eds) => addEdge(connection, eds)), [setEdges] ); return ( <div style={{ width: "100%", height: "600px" }}> <ReactFlow nodes={nodes} edges={edges} onNodesChange={onNodesChange} onEdgesChange={onEdgesChange} onConnect={onConnect} fitView > <Background /> </ReactFlow> </div> ); }

在这个画布上，用户每新建一个节点，后端就把它持久化到CanvasStore。当用户在某节点上点击“提问”，后端就基于这个节点组装上下文，调用 LLM，再把回复作为新节点挂回画布。如此循环，上下文永远以“焦点节点”为中心组织，而不是一地鸡毛的全量记忆。

4.6 串联完整流程

把前面几个模块串起来，写一个命令行演示脚本examples/demo.py：

# examples/demo.py import os from app.canvas_store import CanvasStore from app.context_assembler import ContextAssembler from app.drift_detector import DriftDetector from app.llm_client import LLMClient def run_demo(): store = CanvasStore("demo_canvas.json") # 模拟已有节点：项目背景 + 需求 + 代码片段 project_node = store.add_node( "订单系统是内部管理系统，订单表订单ID、客户ID、订单金额、订单状态四个核心字段。", node_type="doc", x=0, y=0, tags=["项目背景"] ) req_node = store.add_node( "用户需要一个按月统计销售额的报表。", node_type="note", x=150, y=100, parent_id=project_node.node_id, tags=["需求"] ) code_node = store.add_node( "SELECT DATE_TRUNC('month', order_date) AS month, SUM(amount) FROM orders GROUP BY 1;", node_type="code", x=300, y=50, tags=["示例代码"] ) store.add_edge(project_node.node_id, req_node.node_id, "references") store.add_edge(req_node.node_id, code_node.node_id, "depends_on") # 当前用户输入作为新节点 focus_node = store.add_node( "现在请根据上面的表结构和需求，写一个按月统计的 SQL，并加上年份过滤条件。", node_type="chat", x=150, y=300, tags=["当前问题"] ) store.add_edge(focus_node.node_id, project_node.node_id, "references") store.add_edge(focus_node.node_id, code_node.node_id, "references") # 组装上下文 assembler = ContextAssembler(max_tokens=2000) context = assembler.assemble(store, focus_node.node_id) print("=== 组装后的上下文 ===") print(context) # 简单漂移检测（这里没有 embedding，所以跳过） detector = DriftDetector(embed_func=None) drift_flag = detector.is_drift("现在帮我写按月销售额统计SQL", [project_node.content]) print("\n=== 漂移检测 ===") print("是否发生漂移：", drift_flag if drift_flag is not None else "未启用 embedding 检测") # 调用 LLM api_key = os.getenv("OPENAI_API_KEY") if api_key: client = LLMClient(api_key=api_key) answer = client.chat("你是一个 SQL 专家。", context) print("\n=== LLM 回答 ===") print(answer) else: print("\n未检测到 OPENAI_API_KEY，跳过 LLM 调用。") if __name__ == "__main__": run_demo()

运行方式很简单：

python examples/demo.py

如果没有配置 API Key，程序也会先打印“组装后的上下文”，让你直观看到空间画布最终喂给模型的内容是什么。这样即使不花钱调用模型，也能验证上下文管理逻辑是否正确。

5. 运行验证与预期结果

把上面代码跑起来后，你会看到三段输出。

5.1 组装后的上下文

=== 组装后的上下文 === 你是一个基于项目上下文进行回答的助手。 【文档参考】 订单系统是内部管理系统，订单表订单ID、客户ID、订单金额、订单状态四个核心字段。 【对话历史】 现在请根据上面的表结构和需求，写一个按月统计的 SQL，并加上年份过滤条件。

注意这里会出现一个问题：因为对话节点只有一个，它被当作“对话历史”，而真正的问题实际上也是这个节点。当前实现为了演示方便，没有做更精细的区分。实际项目中，你应该把“用户最新输入”单独放一个区块，而不是和旧聊天记录混在一起。

这也是上下文组装器里最容易踩坑的地方：“所有上下文节点”和“用户当前问题”必须严格分离，否则模型会分不清哪些是历史、哪些是要回答的新问题。

5.2 漂移检测

=== 漂移检测 === 是否发生漂移：未启用 embedding 检测

当没有注入 embedding 函数时，检测器直接返回None，这符合预期。如果你想真正体会漂移检测的作用，需要额外实现一个embed_func，这部分依赖于你选择的 embedding 服务，无法在示例代码中统一给出。

5.3 LLM 回答

配置好 API Key 后，你会得到一段 SQL 生成结果。由于我们的上下文里已经包含了“订单金额”字段和月份统计需求，模型大概率会生成类似下面的 SQL：

SELECT TO_CHAR(order_date, 'YYYY-MM') AS month, SUM(order_amount) AS total_amount FROM orders WHERE order_date >= '2022-01-01' GROUP BY TO_CHAR(order_date, 'YYYY-MM') ORDER BY month;

这个结果说明：只要你把正确的字段名和表结构放进上下文，模型就不需要“回忆”之前的对话也能准确完成当前任务。空间画布的价值在这里是可见的。

6. 常见问题与排查思路

把这类系统落地到项目里，通常会遇到以下几个问题。

6.1 节点关联关系建立不对，上下文组装结果混乱

6.2 LLM 调用报错或超时

6.3 token 估算不准导致超限

这是最容易忽视的问题。如果你用前面示例中的简化估算函数，在中文长文本场景下误差可能很大。解决方式有两个：

• 使用官方 tokenizer，例如 OpenAI 的tiktoken库。
• 在发送请求前，把所有消息体交给 tokenizer 估算，如果超限就提前裁剪。

切忌直接在 API 返回context length exceeded后再去裁减，那样会浪费一次无效请求。

6.4 画布数据体积膨胀

节点数量增长后，每次组装都要 BFS 全图，性能会下降。可行的优化方向：

1. 对画布做分区，比如按项目、按会话分组。
2. 在边上加时间戳，组装时优先使用最近活跃的子图。
3. 把组装结果缓存起来，只有焦点节点或相邻节点变化时才重新组装。

6.5 前端拖拽与后端数据不一致

如果使用 React Flow + 后端存储，要注意节点坐标变化时的同步策略。建议在onNodeDragStop事件中调用后端更新接口，而不是每次onNodesChange都提交。否则频繁写库会造成性能问题。

7. 最佳实践与工程建议

7.1 区分“空间画布”与“可视化聊天记录”

需要明确一点：空间节点画布不是普通的聊天记录展示器。它真正要做的是为 LLM 生成上下文提供结构化的信息来源。如果你只是把聊天记录画在画布上，但不改变组装逻辑，那 context drift 问题并不会自动消失。

一个有效的评判标准是：当用户从节点 A 跳到节点 B 提问时，模型看到的上下文是否完全不同。如果两次提问都拿到的是同一大坨历史，你的画布就只是“看起来酷炫”的 UI，而不是一个上下文管理器。

7.2 上下文预算要显式管理

推荐在系统里引入“三层上下文优先级”：

1. 系统层：角色设定、任务规则、全局约束。始终保留。
2. 应用层：当前任务相关的文档、代码、决策记录。根据相关性排序，优先保留。
3. 会话层：历史对话。只保留与当前焦点关联最近的少量轮次，必要时用摘要代替原文。

这套设计与 RAG 的思想有相似之处，但 RAG 解决的是“从外部知识库检索缺失知识”，空间画布解决的是“在内部已有信息中找回焦点”。两者可以叠加使用：画布节点本身也可以是 RAG 检索回来的文档片段。

7.3 漂移检测的阈值需要数据支撑

我在第 4.2 节给出了一个简单的相似度阈值方案。实际项目中，建议提前收集一批“正常对话”和“漂移对话”样本，计算它们的相似度分布，然后用统计结果确定阈值。不要拍脑袋定一个 0.7 或 0.5。

如果这个环节做得够好，你可以进一步实现自动建议：当检测到用户输入与当前焦点主题差异过大时，前端自动提示“是否要新建一个话题分支”，而不是把新问题硬塞进当前上下文。这才是空间画布相比普通滑动窗口对话的最大优势。

7.4 安全与合规边界

接入 LLM 时，以下几点需要重视：

• API Key 管理：使用 environment variable 或密钥管理服务，不要硬编码到代码库。
• 敏感信息脱敏：不要把用户手机号、身份证号等个人敏感信息直接写入上下文节点。画布数据通常会被持久化，一旦泄露影响范围更大。
• 输出校验：LLM 返回的代码片段不能直接执行，尤其是涉及数据库操作时，需要先经过人工确认或安全扫描，避免生成带有破坏性的 SQL 或命令。
• 合法授权：如果系统会读取用户私有知识库或企业文档，需要确保已获得相应授权。

7.5 设计上预留“降级回退”通道

空间画布是个不错的上下文管理方案，但并不是所有用户都愿意手动拖拽节点、建立连线。建议保留一个“自动模式”：当用户不主动维护画布时，系统自动把会话转换成节点并做常规连接；等到用户觉得回答质量下降时，再打开画布手动调整关联关系。这样既保证了基本体验，也给了用户一个干预入口。

8. 总结与后续学习路线

本文从 LLM 上下文漂移（context drift）这个实际问题出发，介绍了空间节点画布（spatial node canvas）的基本思路，并给出了一个可运行的 Python 原型。整个系统的核心不在于画布 UI，而在于：

1. 把上下文拆成可管理、可关联的节点；
2. 以当前焦点节点为中心做图遍历，筛选相关节点；
3. 按类型和优先级排序，并显式控制 token 预算；
4. 用结构化上下文替代流水账式历史，从源头降低上下文漂移风险。

如果你接下来想继续深入，建议按以下路线学习：

• 先把你手头的对话系统改成“显式上下文组装”模式，即使没有画布 UI，只把历史记录分成“背景文档、代码参考、对话记录、当前问题”四段，效果也会比纯拼接好很多。
• 再学习图数据库相关概念，了解如何用图结构表达更复杂的上下文关系，而不是停留在简单的 BFS 遍历阶段。
• 然后研究 RAG 与画布的融合，让画布节点能动态引用外部知识库内容。
• 最后把漂移检测做扎实，用真实业务数据持续调优阈值。

上下文管理的核心原则很简单：模型能看到什么，决定了它能回答成什么样。与其给模型更长窗口去“大海捞针”，不如用工程手段主动把针递到模型手里。空间节点画布只是实现这个目标的一种交互形态，背后的结构化上下文组装思想，才是真正值得沉淀到项目里的东西。

如果你在实现过程中遇到上下文组装顺序混乱、token 超限或者画布数据持久化的问题，欢迎带着具体场景继续交流。动手把这个 demo 跑起来，你会对 LLM 应用的上下文管理有一个更直观的理解。