Agent 工程体系 · 第 27/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Agent 会话隔离：用户、租户、线程、群聊成员和上下文串线

Agent 的“记忆”不是一个单一对象，而是多个作用域不同的状态集合：某个用户的偏好、某个租户的知识库、某个线程中的历史消息、群聊当前可见成员，以及本次请求临时装配的上下文。

如果这些作用域被错误地合并，系统就会出现上下文串线：用户 A 看到用户 B 的信息，租户甲的知识进入租户乙的回答，群聊成员获得不应看到的私聊内容，或者一个线程中的工具结果影响了另一个线程。

这类问题不是“模型记性太好”，而是系统把不属于当前主体的状态放进了模型输入、工具参数、检索结果或持久化存储。

1. 先区分五个概念

1.1 用户：谁在提出请求

用户是现实中的主体，通常由认证系统识别，例如：

user_id = u_123

用户身份回答的是：

这条消息是谁发出的？

用户不等于登录会话，也不等于线程。

同一个用户可以同时存在多个线程：

u_123
├── thread_personal_finance
├── thread_project_alpha
└── thread_customer_support

因此，用户级状态适合保存跨线程仍然成立的信息，例如：

{
  "user_id": "u_123",
  "preferences": {
    "language": "zh-CN",
    "address_user_as": "老王",
    "timezone": "Asia/Shanghai"
  }
}

但“用户正在处理项目 Alpha”通常不应直接作为永久用户画像。它可能只是某个线程的临时事实。

1.2 租户：数据和权限属于哪个组织边界

租户是多租户系统中的隔离主体，常见于企业、工作区、客户组织或项目空间：

tenant_id = tenant_acme

租户回答的是：

这条请求允许访问哪个组织的数据、模型、工具和资源？

同一个用户可能属于多个租户：

user_id = u_123
├── tenant_acme
└── tenant_beta

因此，仅凭 user_id 不能决定数据访问范围。查询条件至少需要同时包含租户：

SELECT *
FROM documents
WHERE tenant_id = :tenant_id
  AND document_id = :document_id;

错误写法是：

SELECT *
FROM documents
WHERE document_id = :document_id;

如果 document_id 可猜测，或者应用层误传了别的租户的 ID，这条查询就可能越过租户边界。

租户与用户的关系是授权关系，而不是字符串拼接关系。应显式验证：

membership(user_id, tenant_id) = active

并进一步验证用户在该租户中的角色、资源权限和工具权限。

1.3 线程：一条连续任务或对话的状态边界

线程，也可称为 conversation、session 或 workflow instance，是一组具有连续因果关系的交互状态：

thread_id = th_456

线程回答的是：

这条消息应该接在哪一段历史之后？

线程通常包含：

消息历史；
工具调用及工具返回；
当前任务状态；
人工审批状态；
中间推理所需的结构化数据；
尚未完成的工作流节点。

OpenAI 的 Responses API 支持通过 previous_response_id 把一次响应连接到下一次响应，从而形成有顺序的响应链；也可以使用具有持久标识的 Conversations API 对象跨会话、设备或任务保存会话状态。Conversation 中保存的不只是文本消息，还可以包括工具调用和工具输出。(developers.openai.com)

线程不是用户身份：

user_id = u_123
thread_id = th_456

两者应分别存储、分别校验：

thread.owner_user_id == authenticated_user_id
thread.tenant_id       == authenticated_tenant_id

只传 thread_id 而不验证归属，是典型的越权入口。

1.4 群聊成员：当前消息可见的主体集合

私聊通常有一个主要用户，但群聊的请求主体是一个集合：

members = {u_alice, u_bob, u_cathy}

群聊成员回答的是：

当前 Agent 的回答应该对哪些人可见？哪些人的信息可以参与当前上下文？

群聊中的“可见”至少有三层：

消息可见性：成员能否看到这条消息；
资料可见性：成员能否看到其他成员的姓名、职位、偏好或历史发言；
资源可见性：Agent 能否代表成员访问文件、日历、工单或外部系统。

这三层不能混为一谈。

例如，Alice 在群里说：

“我刚才在私聊中告诉 Agent，我下周要休假。”

即使 Agent 能从自己的用户记忆中取到 Alice 的休假信息，也不能因此把它自动加入群聊上下文。用户级记忆可被当前用户调用，不代表可被群聊中的其他成员调用。

1.5 上下文：真正送进模型或工具的数据集合

上下文不是“聊天记录”的同义词，而是一次模型调用时实际可见的输入：

context =
    system_instructions
  + current_user_message
  + thread_history
  + selected_user_memory
  + selected_tenant_knowledge
  + visible_group_messages
  + tool_results
  + runtime_metadata

上下文包括显式文本，也包括结构化信息：

{
  "tenant_id": "tenant_acme",
  "user_id": "u_alice",
  "roles": ["billing.read"],
  "locale": "zh-CN",
  "retrieved_documents": ["doc_1", "doc_9"],
  "tool_results": [
    {
      "tool": "calendar.search",
      "data": "Alice 下周一休假"
    }
  ]
}

模型无法自动知道这些字段是否应该出现。只要应用把数据放入请求，模型通常就会把它视为当前任务的一部分。

因此，隔离的核心不是“提示模型不要泄露”，而是建立一个更强的条件：

不属于当前授权作用域的数据，不进入模型上下文，也不进入当前工具调用。

2. 会话隔离的形式化条件

设一次请求为：

r = (tenant_id, actor_user_id, channel_id, thread_id, message_id)

其中：

tenant_id：当前租户；
actor_user_id：实际发消息的用户；
channel_id：私聊、群聊、机器人频道等通信空间；
thread_id：连续对话或工作流实例；
message_id：当前消息的唯一标识。

设系统中的状态分为：

S = {
  S_user,
  S_tenant,
  S_thread,
  S_channel,
  S_message,
  S_runtime
}

可以把每类状态定义为不同作用域：

scope(S_user)   = user_id
scope(S_tenant) = tenant_id
scope(S_thread) = (tenant_id, thread_id)
scope(S_channel)= (tenant_id, channel_id)
scope(S_message)= (tenant_id, channel_id, message_id)

一次请求真正允许使用的上下文应满足：

C(r) \subseteq A(r)

其中：

$C(r)$ 是实际装配的上下文；
$A(r)$ 是当前请求经认证和授权后允许访问的数据集合。

进一步，对任何一条被放入上下文的数据 $x$ ，都应满足：

\text{scope}(x) \cap \text{authority}(r) \neq \varnothing

直觉是：数据的作用域必须与当前请求的授权范围相交。

例如：

当前请求：
tenant_id = acme
actor_user_id = alice
thread_id = t1
group_members = {alice, bob}

则：

Alice 的语言偏好       可以使用
t1 的消息历史          可以使用
Acme 的公开知识库      需通过租户权限过滤后使用
Alice 的私聊记忆       不能直接对 Bob 展示
Beta 租户的文档        不可以使用
t2 的未授权历史        不可以使用

2.1 “可读取”与“可展示”不是同一个权限

设：

$R(r, x)$ ：请求 $r$ 是否可以读取数据 $x$ ；
$D(r, x)$ ：请求 $r$ 是否可以将数据 $x$ 展示给当前可见成员；
$M(r, x)$ ：数据 $x$ 是否进入模型上下文。

安全条件不是只有：

R(r, x) = 1

而是：

M(r, x) = 1 \Rightarrow R(r, x) = 1

对于群聊或多人可见场景，还要满足：

D(r, x) = 1

因此：

Agent 可以从 Alice 的私有日历读取信息

不等于：

Agent 可以把该信息放进 Alice 和 Bob 都能看到的群聊回答

这是群聊串线中最容易被忽略的因果差异。

3. 五种作用域应如何组合

实际系统通常至少需要下表中的五种状态。

状态	典型内容	默认作用域	是否跨线程
用户状态	称呼、语言、确认过的偏好	用户	是
租户状态	租户知识、策略、配额、工具配置	租户	是
线程状态	消息、工具结果、任务进度	租户 + 线程	否
群聊状态	成员、角色、可见消息、频道设置	租户 + 群组	视设计
请求上下文	当前消息、授权结果、检索结果	单次请求	否

关键不是把所有数据都“隔离”，而是明确哪些状态允许共享。

例如：

用户偏好：
  可跨线程，但必须属于同一 user_id

租户知识库：
  可跨用户、跨线程，但必须属于同一 tenant_id 且通过权限过滤

线程历史：
  只能属于同一个 thread_id

群聊消息：
  只能对当前频道成员可见

请求上下文：
  请求结束后应丢弃，除非经过显式提炼并写入某个长期作用域

最后一条很重要。模型在当前线程中推断出的内容，不应自动升级为用户长期记忆：

用户说：“这次项目先用英文回复。”

更合理的解释是：

thread_preference.language = "en"

而不是：

user_preference.language = "en"

除非用户明确表达：

“以后都用英文回复。”

4. 会话状态 API 不等于授权系统

OpenAI 的会话状态能力解决的是“如何保存和延续上下文”，不自动解决“谁有权访问这个上下文”。

例如，使用 previous_response_id：

from openai import OpenAI

client = OpenAI()

first = client.responses.create(
    model="gpt-5.6",
    input="我正在处理项目 Alpha。",
)

second = client.responses.create(
    model="gpt-5.6",
    previous_response_id=first.id,
    input="刚才那个项目的截止日期是什么？",
)

print(second.output_text)

第二次请求通过 previous_response_id 延续第一次响应链。这个链条能表达线程连续性，但应用仍然需要在进入请求前确认：

first.id 是否确实属于当前 user_id、tenant_id 和 thread_id？

如果服务端把任意客户端提交的响应 ID 原样作为 previous_response_id，就可能出现：

Alice 提交 Bob 的 response_id
→ 服务端继续 Bob 的上下文
→ 模型看到 Bob 的历史
→ Alice 得到 Bob 的信息

所以，响应 ID 或 conversation ID 应被视为不透明的状态引用，而不是授权凭证。

数据库中至少应有一张归属表：

CREATE TABLE conversations (
    conversation_id TEXT PRIMARY KEY,
    tenant_id       TEXT NOT NULL,
    owner_user_id   TEXT,
    channel_id      TEXT,
    thread_id       TEXT NOT NULL,
    visibility      TEXT NOT NULL,
    created_at      TIMESTAMP NOT NULL
);

CREATE UNIQUE INDEX conversations_tenant_thread
ON conversations (tenant_id, thread_id);

读取时必须带上授权条件：

SELECT conversation_id
FROM conversations
WHERE conversation_id = :conversation_id
  AND tenant_id = :tenant_id
  AND (
      owner_user_id = :actor_user_id
      OR visibility = 'group'
  );

更严格的实现还应继续验证群组成员关系，而不是把 visibility = 'group' 当成所有用户都可见。

OpenAI 文档还区分了响应对象和 Conversation 对象的持久化行为：响应对象默认保存 30 天，而附着在 Conversation 上的项目不受该 30 天 TTL 约束；即使使用 previous_response_id，链中此前的输入 token 仍会计入输入计费。(developers.openai.com)

这带来两个工程边界：

状态生命周期必须由应用根据隐私、合规和业务需要设计；
成本控制不能简单认为“引用前一个响应就不需要重复处理历史”。

5. LangGraph 中的 thread 与跨线程 Store

LangGraph 将短期线程状态和长期跨线程数据明确区分：

Checkpointer 保存图状态快照，作用域是单个 thread；
Store 保存应用定义的键值数据，可跨 thread 使用；
thread_id 通过图配置传入，用于恢复对应线程状态。(docs.langchain.com)

一个最小示例：

from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore


class State(TypedDict):
    messages: Annotated[list, add_messages]


def assistant(state: State):
    last_message = state["messages"][-1]
    text = last_message.content

    return {
        "messages": [
            {
                "role": "assistant",
                "content": f"收到：{text}",
            }
        ]
    }


builder = StateGraph(State)
builder.add_node("assistant", assistant)
builder.add_edge(START, "assistant")

checkpointer = InMemorySaver()
store = InMemoryStore()
graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)

config_alice = {
    "configurable": {
        "thread_id": "tenant-acme:user-alice:thread-1"
    }
}

result1 = graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "我正在处理项目 Alpha。"}
        ]
    },
    config_alice,
)

result2 = graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "我刚才在处理什么项目？"}
        ]
    },
    config_alice,
)

print(result2["messages"][-1].content)

两次调用使用同一个 thread_id，因此第二次可以恢复线程状态。若改成：

config_bob = {
    "configurable": {
        "thread_id": "tenant-acme:user-bob:thread-1"
    }
}

则它是另一个线程，不应看到 Alice 的消息。

但需要注意：thread_id 只是框架定位状态的键，不是天然安全边界。以下写法仍然有风险：

thread_id = request.json["thread_id"]
result = graph.invoke(input_state, {
    "configurable": {"thread_id": thread_id}
})

正确流程应是：

thread = load_thread(thread_id)

if thread is None:
    raise NotFound()

if thread.tenant_id != authenticated_tenant_id:
    raise Forbidden()

if not can_access_thread(
    user_id=authenticated_user_id,
    thread=thread,
):
    raise Forbidden()

result = graph.invoke(
    input_state,
    {
        "configurable": {
            "thread_id": thread.internal_thread_key
        }
    },
)

这里的关键顺序是：

认证 → 租户确定 → 线程归属验证 → 加载线程状态 → 调用 Agent

而不是：

先按客户端给的 thread_id 加载状态 → 再尝试判断用户是谁

LangGraph 文档还指出，MemorySaver 或 InMemorySaver 将检查点放在内存中，进程重启后会丢失；生产环境应使用持久化 checkpointer。长对话还会导致检查点持续累积，需要通过保留策略或定期清理控制存储和延迟。(docs.langchain.com)

这些问题不一定造成串线，但会造成另一类状态错误：

重启后线程历史消失
→ Agent 误以为任务从未开始
→ 重复执行工具或重复请求人工审批

因此，持久化可靠性和隔离正确性是两个不同维度，不能互相替代。

6. 用户记忆不应直接进入群聊

考虑如下数据：

用户 Alice 的长期记忆：
{
  "user_id": "alice",
  "fact": "Alice 正在寻找新的工作机会"
}

Alice 在群聊 team-1 中询问：

“帮我们安排下周的项目会议。”

Agent 如果检索到 Alice 的长期记忆，并生成：

“既然 Alice 正在找工作，我们把会议安排在她离职前吧。”

即使这条记忆确实属于 Alice，仍然发生了隐私泄露，因为回答的可见对象包括其他成员。

正确的上下文装配应将“用户可读取”和“群聊可展示”分开：

def build_context(request, authz):
    visible_messages = load_visible_channel_messages(
        tenant_id=request.tenant_id,
        channel_id=request.channel_id,
        member_ids=authz.visible_member_ids,
    )

    thread_state = load_thread_state(
        tenant_id=request.tenant_id,
        thread_id=request.thread_id,
    )

    user_memory = load_user_memory(
        tenant_id=request.tenant_id,
        user_id=request.actor_user_id,
    )

    if request.channel_type == "group":
        displayable_memory = [
            item for item in user_memory
            if item.visibility == "group"
            and authz.can_disclose(item)
        ]
    else:
        displayable_memory = user_memory

    return {
        "messages": visible_messages,
        "thread_state": thread_state,
        "user_memory": displayable_memory,
    }

但即使用户记忆设置为 group，也不能只靠字段判断。还应考虑：

该成员是否仍在群中；
记忆是否过期；
该群是否有额外的敏感信息规则；
当前 Agent 是否具有展示该类信息的权限；
其他成员是否有查看该字段的权限。

一种更安全的数据结构是把“事实”和“披露策略”分开：

{
  "memory_id": "m_901",
  "subject_user_id": "alice",
  "fact": "偏好使用中文",
  "source": "explicit_user_statement",
  "visibility": "user_only",
  "allowed_channels": [],
  "expires_at": null
}

{
  "memory_id": "m_902",
  "subject_user_id": "alice",
  "fact": "在项目会议中使用中文",
  "source": "explicit_user_statement",
  "visibility": "channel",
  "allowed_channels": ["team-1"],
  "expires_at": "2026-12-31T00:00:00Z"
}

来源可信度和可见范围都需要单独建模。模型猜测出的内容不能自动获得高可信度和广泛可见性。

7. 租户隔离必须覆盖整条数据流

很多系统只在主数据库查询时过滤 tenant_id，但上下文串线往往发生在其他组件。

一次 Agent 请求的数据流可能是：

flowchart LR
    A[认证请求] --> B[身份与租户解析]
    B --> C[线程归属校验]
    C --> D[权限计算]
    D --> E[消息与线程状态]
    D --> F[用户记忆]
    D --> G[租户知识检索]
    D --> H[工具凭证选择]
    E --> I[上下文装配]
    F --> I
    G --> I
    H --> J[工具调用]
    J --> I
    I --> K[模型]
    K --> L[输出过滤]
    L --> M[当前频道]

租户边界必须在这些位置同时成立：

数据库

SELECT id, title, content
FROM documents
WHERE tenant_id = :tenant_id
  AND embedding <=> :query_embedding < :threshold;

向量库

向量检索的过滤条件不能只依赖应用层：

{
  "vector": [0.01, 0.02, 0.03],
  "top_k": 8,
  "filter": {
    "tenant_id": "acme",
    "visibility": "internal"
  }
}

如果先检索全库、再在应用层过滤，错误的文档可能已经：

进入日志；
进入缓存；
进入重排模型；
进入模型上下文；
出现在调试追踪中。

缓存

错误的缓存键：

rag:query:如何申请发票

正确的缓存键至少需要包含租户、权限版本和知识库版本：

rag:{tenant_id}:{permission_hash}:{kb_version}:{query_hash}

否则租户甲的检索结果可能被租户乙命中。

工具和密钥

工具调用不能只传：

{
  "customer_id": "c_123"
}

还应由服务端根据当前授权上下文决定租户和凭证：

{
  "tenant_id": "acme",
  "actor_user_id": "alice",
  "customer_id": "c_123"
}

工具服务再次校验：

customer.tenant_id == request.tenant_id

不应允许模型自由传入任意 tenant_id，更不能让模型选择任意租户的 API Key。

日志和追踪

如果日志记录了完整 prompt，则日志系统本身也成为数据副本。日志至少需要：

{
  "request_id": "req_001",
  "tenant_id": "acme",
  "actor_user_id": "alice",
  "thread_id": "th_1",
  "context_item_ids": ["msg_1", "doc_7"],
  "redaction": "applied"
}

生产日志不应默认记录：

全量私聊内容；
其他成员的私人记忆；
API Key；
工具返回的完整敏感字段；
未脱敏的检索片段。

8. 典型串线案例：错误的共享内存

下面的代码看起来很方便，但会把所有用户写入同一个全局列表：

history = []

def handle_message(user_id, text):
    history.append({"role": "user", "content": text})
    answer = call_model(history)
    history.append({"role": "assistant", "content": answer})
    return answer

并发请求时，可能出现以下交错：

请求 A：history.append("Alice：我的工资是……")
请求 B：history.append("Bob：帮我总结一下")
请求 B：call_model(history)

于是 Bob 的模型输入包含 Alice 的内容。

问题不是 Python 列表本身，而是状态键缺失。至少需要按线程分区：

histories = {}

def handle_message(tenant_id, thread_id, text):
    key = (tenant_id, thread_id)
    history = histories.setdefault(key, [])

    history.append({"role": "user", "content": text})
    answer = call_model(history)
    history.append({"role": "assistant", "content": answer})
    return answer

但这仍然不是完整方案，因为：

没有验证调用者是否能访问该线程；
多进程部署时不同实例各自拥有不同历史；
并发请求可能同时读写同一线程；
进程重启后状态丢失；
群聊成员可见性仍未处理；
工具结果可能写入错误的线程。

所以，正确的线程键通常不是单独的字符串，而是经过服务端确认的复合身份：

ThreadKey = (tenant_id, channel_id, thread_id)

应用可以把它编码为一个内部键：

import hashlib


def internal_thread_key(tenant_id, channel_id, thread_id):
    raw = f"{tenant_id}\x00{channel_id}\x00{thread_id}"
    return hashlib.sha256(raw.encode()).hexdigest()

哈希只能避免键过长或暴露原始信息，不能替代权限检查。LangGraph 文档也提示，某些持久化实现对 thread_id 有长度限制，超过 255 个字符可能产生数据库错误；需要确定性 ID 时可使用 UUID 或哈希。(docs.langchain.com)

9. 并发下的线程状态不能靠“最后写入获胜”

即使每个线程都有独立键，同一线程的并发仍然会造成状态损坏。

假设线程 t1 当前版本为 v10：

v10 = [用户：创建报表]

两个请求同时到达：

请求 A 读取 v10
请求 B 读取 v10

A 追加：筛选华东区域
B 追加：按月汇总

A 写入 v11
B 写入 v11

如果数据库没有版本控制，B 可能覆盖 A：

最终 v11 = [创建报表, 按月汇总]

“筛选华东区域”丢失。

可使用乐观并发控制：

UPDATE thread_state
SET state_json = :new_state,
    version = version + 1
WHERE tenant_id = :tenant_id
  AND thread_id = :thread_id
  AND version = :expected_version;

检查影响行数：

affected_rows = 1 → 写入成功
affected_rows = 0 → 版本冲突，需要重新加载或拒绝请求

对于具有副作用的工具调用，还需要幂等键：

idempotency_key = hash(tenant_id, thread_id, message_id, tool_name)

否则重试路径可能造成：

第一次调用：创建订单成功
网络超时
第二次重试：再次创建订单

线程隔离保证“写到了正确的线程”，幂等和版本控制保证“没有把正确线程写坏”。

10. 群聊必须定义“当前发言人”和“可见成员”

群聊请求不能只表示成：

{
  "thread_id": "t1",
  "text": "帮我查一下进度"
}

更完整的请求应包含：

{
  "tenant_id": "acme",
  "channel_id": "team-1",
  "thread_id": "t1",
  "actor_user_id": "alice",
  "visible_member_ids": ["alice", "bob", "cathy"],
  "message_id": "msg-88",
  "text": "帮我查一下进度"
}

服务端不能完全信任客户端提交的 visible_member_ids，而应从频道成员关系和消息平台事件中重新计算：

def resolve_group_scope(channel_id, actor_user_id, message_id):
    channel = load_channel(channel_id)
    message = load_message(message_id)

    if message.channel_id != channel_id:
        raise Forbidden()

    if actor_user_id not in channel.active_member_ids:
        raise Forbidden()

    return {
        "channel_id": channel_id,
        "actor_user_id": actor_user_id,
        "visible_member_ids": channel.active_member_ids,
    }

在上下文装配时，应把成员来源标记清楚：

{
  "role": "user",
  "author_id": "alice",
  "visibility": "channel",
  "content": "帮我查一下进度"
}

不要把多人的消息全部转换成：

{
  "role": "user",
  "content": "Alice：...\nBob：...\nCathy：..."
}

因为这样会丢失作者、可见范围和消息来源。后续的记忆提炼、引用和权限判断就无法准确进行。

11. 记忆提炼是跨作用域升级操作

上下文串线不只发生在“读取”时，也可能发生在“写入记忆”时。

例如，群聊中 Bob 说：

“我们下季度暂时不用 Redis。”

Agent 如果把它写入 Bob 的长期用户画像：

{
  "user_id": "bob",
  "fact": "Bob 不使用 Redis"
}

就产生了两个错误：

这可能是团队项目决策，不是 Bob 的个人偏好；
原始内容属于群聊上下文，不应自动变成 Bob 的跨线程个人事实。

记忆提炼应先判断目标作用域：

消息来源作用域       候选记忆作用域
私聊中的明确个人偏好 → 用户记忆
群聊中的项目决策     → 租户或项目记忆
当前任务中的临时事实 → 线程状态
工具返回的瞬时数据   → 不一定持久化

可以将提炼结果设计为带作用域的事件：

{
  "candidate": "下季度项目暂时不用 Redis",
  "source": {
    "tenant_id": "acme",
    "channel_id": "team-1",
    "thread_id": "t1",
    "message_id": "msg-88"
  },
  "proposed_scope": "project",
  "confidence": 0.91,
  "requires_confirmation": true
}

只有在规则允许、来源可信且作用域明确时，才写入长期存储：

def promote_memory(candidate, request):
    if candidate.proposed_scope == "user":
        if candidate.subject_user_id != request.actor_user_id:
            raise PolicyError("不能代替其他用户写入个人记忆")

    if candidate.proposed_scope in {"tenant", "project"}:
        if request.channel_type not in {"group", "project"}:
            raise PolicyError("缺少组织上下文")

    if candidate.requires_confirmation:
        return "pending_confirmation"

    persist_memory(candidate)
    return "saved"

“模型觉得这件事重要”不是写入权限。“信息出现过”也不是永久记忆资格。

12. 诊断上下文串线：先定位泄漏发生在哪一层

出现疑似串线时，不要只检查最终回答。应沿着数据流逐层检查。

12.1 先确认请求身份

记录但不直接暴露敏感内容：

{
  "request_id": "req-100",
  "authenticated_user_id": "alice",
  "authenticated_tenant_id": "acme",
  "channel_id": "team-1",
  "thread_id": "t1",
  "message_id": "m1"
}

重点检查：

actor_user_id 是否来自可信认证上下文？
tenant_id 是否由服务端解析？
thread_id 是否通过归属校验？
channel_id 是否与消息一致？

12.2 再检查上下文来源集合

为每一项上下文生成来源记录：

{
  "context_item_id": "ctx-7",
  "source_type": "memory",
  "source_id": "mem-22",
  "tenant_id": "acme",
  "owner_user_id": "bob",
  "visibility": "user_only",
  "included_for": ["alice"]
}

如果最终输入包含：

owner_user_id = bob
included_for = alice
visibility = user_only

则无需等模型输出泄露，装配阶段已经失败。

12.3 检查检索和缓存

对每个检索结果记录：

query_tenant_id
result_tenant_id
query_user_id
required_permission
actual_permission
cache_key

常见错误是：

查询使用 tenant_id = acme
缓存命中来自 tenant_id = beta

或者：

数据库查询已过滤租户
向量检索没有过滤租户

12.4 检查工具调用

工具日志应能回答：

谁调用了工具？
代表哪个租户？
使用哪条凭证？
访问了哪个资源？
资源属于哪个租户？
工具是否执行了二次授权？

仅记录：

tool=search_customer, customer_id=c123

无法判断是否跨租户。

12.5 区分“模型幻觉”和“真实串线”

如果回答出现另一个用户的信息，可能有三种来源：

上下文确实包含了该信息；
工具返回了该信息；
模型自行生成了看似具体但并不存在的信息。

应保存可审计的上下文项 ID，而不是只保存完整 prompt。这样可以判断：

泄露内容是否能在 context_item_id 对应的数据中找到？

若能找到，优先排查隔离和授权；若找不到，再排查模型生成和输出校验。

13. 反例：看似安全但仍然错误的设计

反例一：用 user_id 作为唯一线程键

thread_id = user_id

后果是用户的所有任务共享一条历史：

财务咨询 → 公司招聘 → 家庭旅行

这些任务之间没有自然的因果连续性，模型会把无关信息混在一起。

正确做法是：

thread_id 独立生成
owner_user_id 记录归属

反例二：用 thread_id 作为全部授权依据

if thread_id:
    load_thread(thread_id)

线程 ID 能定位数据，但不能证明调用者有权限访问数据。必须查询线程归属、租户归属和成员关系。

反例三：把 tenant_id 拼进 prompt 就认为完成隔离

system = f"当前租户是 {tenant_id}，不要访问其他租户。"

这只是给模型的指令，不会阻止：

数据库查询跨租户；
向量库返回错误租户；
缓存命中其他租户；
工具服务接受任意租户参数；
日志保存别的租户信息。

租户隔离必须发生在数据访问层和工具执行层，prompt 中的租户信息只能作为辅助上下文。

反例四：把整个用户记忆注入每次请求

context["user_memory"] = load_all_memory(user_id)

这会导致：

当前任务看到无关信息；
群聊泄露私密信息；
过期事实影响回答；
敏感属性进入不必要的模型调用；
token 成本和上下文噪声增加。

应该按当前任务、可见范围、数据新鲜度和权限选择记忆。

反例五：把共享 Store 当作线程状态

LangGraph 中，Store 适合保存跨线程的用户偏好、事实和共享知识；Checkpointer 才用于单线程图状态。(docs.langchain.com)

如果把当前任务中的临时变量写入跨线程 Store：

store.put(
    ("users", user_id),
    "current_plan",
    {"step": 3, "draft": "..."}
)

用户在另一个线程中就可能看到上一任务的草稿。

应根据数据语义选择：

当前任务进度 → checkpointer / thread state
用户长期偏好 → user-scoped store
团队共享知识 → tenant/project-scoped store

14. 一套可验证的隔离不变量

生产系统应把隔离要求写成可测试的不变量，而不是只写在设计文档中。

不变量一：线程归属

每次加载 thread_state，都必须满足：
thread.tenant_id == request.tenant_id
且 actor 对 thread 有访问权限

不变量二：用户记忆归属

user_memory.user_id == request.actor_user_id

除非存在明确的代理授权，例如管理员代表用户操作。

不变量三：租户资源归属

每个 document、tool_resource、credential、cache_entry
都必须能追溯到 request.tenant_id

不变量四：群聊展示范围

output 中的每条事实，都必须允许对当前 visible_member_ids 展示

不变量五：跨作用域写入必须显式升级

thread → user
thread → tenant
user → group

这些都不是普通写操作，而是作用域升级。应经过单独策略判断、审计和必要的用户确认。

15. 测试必须模拟“交叉组合”，而不是只测单用户

单元测试只创建一个用户和一个线程，无法发现串线。至少应构造如下测试矩阵：

租户：
  acme、beta

用户：
  alice、bob

线程：
  t_alice_private
  t_bob_private
  t_team_group

资源：
  doc_acme、doc_beta

记忆：
  mem_alice_private
  mem_bob_private
  mem_team_shared

然后验证：

def test_tenant_isolation():
    answer = ask(
        tenant_id="acme",
        user_id="alice",
        thread_id="t_alice_private",
        text="搜索公司报销制度",
    )

    assert "doc_beta" not in answer.context_source_ids


def test_thread_isolation():
    answer = ask(
        tenant_id="acme",
        user_id="alice",
        thread_id="t_alice_private",
        text="我刚才在处理什么？",
    )

    assert "t_bob_private" not in answer.context_source_ids


def test_group_does_not_expose_private_memory():
    answer = ask(
        tenant_id="acme",
        user_id="alice",
        channel_id="team-1",
        thread_id="t_team_group",
        text="总结一下我的相关信息",
    )

    assert "mem_alice_private" not in answer.context_source_ids

还需要测试并发：

Alice 和 Bob 同时请求
同一个线程的两个消息同时请求
工具调用超时后重试
服务进程重启后恢复线程
用户退出群聊后访问旧消息
用户从租户甲切换到租户乙
缓存预热和缓存命中

测试目标不是只看回答文本，而是检查：

实际模型输入
实际检索结果
实际工具参数
实际凭证
实际持久化键
实际输出可见成员

16. 生产设计的最小闭环

一个请求可以按以下顺序处理：

1. 从认证上下文获得 actor_user_id
2. 确定当前 tenant_id，不接受任意客户端租户切换
3. 校验用户是否属于该租户
4. 校验 channel_id 和 message_id 的关系
5. 校验 actor 是否属于当前群聊
6. 校验 thread_id 是否属于当前租户和频道
7. 计算当前可见成员集合
8. 查询线程状态
9. 查询经过权限过滤的用户记忆和租户知识
10. 选择当前租户允许使用的工具和凭证
11. 装配带来源和可见范围的上下文
12. 调用模型或 Agent 图
13. 对工具副作用执行幂等和二次授权
14. 对输出执行可见范围检查
15. 仅将明确允许的数据写入对应作用域
16. 记录可审计的来源 ID 和授权决策

其中最不能省略的是第 11 步和第 14 步：

上下文装配决定模型“能看到什么”；
输出检查决定成员“能看到什么”。

仅做输出过滤会太晚：敏感数据可能已经被模型推理、工具调用或日志记录。仅做输入过滤也不够：模型可能生成新的敏感内容，或者工具返回了超出当前可见范围的信息。

17. 最后的判断标准

判断一个 Agent 会话是否真正隔离，不应问：

我们有没有给每个用户一个 session ID？

而应逐项回答：

当前请求由谁发起？
属于哪个租户？
接在哪个线程？
当前频道有哪些成员？
每条上下文来自哪个作用域？
每个检索结果属于哪个租户？
每个工具调用使用谁的权限和密钥？
模型输出将对谁可见？
当前事实会写入哪个长期记忆作用域？
并发重试是否会覆盖或重复执行？
审计记录能否还原数据为什么被使用？

用户、租户、线程和群聊成员解决的是不同的身份与范围问题；上下文则是这些边界最终汇合的地方。只要任意一个边界在装配上下文之前没有被明确验证，系统就不能证明自己没有串线。

OpenAI 的 Conversation/Responses 状态机制可以帮助保存和延续会话，LangGraph 的 Checkpointer 与 Store 可以分别承载线程级和跨线程状态，但这些机制本身都不替代应用的身份绑定、租户授权、成员可见性和数据来源审计。(developers.openai.com)

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。