在一个托管智能体会话中复用多名用户

#在一个托管智能体会话中复用多名用户

默认情况下，每个调用方都有自己的托管智能体（hosted agent）会话，如《按用户隔离托管智能体会话》所述。服务很多用户的应用——一个 Teams 机器人、一个独立软件供应商（ISV）网关，或一个客户支持平台——并不需要每个用户一个会话。中间层服务把很多用户映射到一个有上限的共享会话池，并在每次调用时标明每个用户。

本文说明如何从你的中间层把会话跨用户做成池，同时让每个用户的数据在共享会话内部保持隔离。

即便用户共用一个会话，平台也会为你隔离对话状态：一个用户创建的响应链，另一个用户不能通过 previous_response_id 接着往下走；context.get_history() 只返回当前请求所属用户被授权看到的历史。你负责两件事：中间层里从用户到会话的映射，以及容器自己存储的任何数据（文件、行或缓存）如何分区。这些数据超出了平台管理的对话状态。

一份完整、可运行的会话复用（session multiplexing）示例演示了两侧：中间层会话池和容器处理函数。本文在往下讲时会链到它的文件。

先决条件

• 一个使用容器协议版本 2.0.0 的托管智能体。要升级，见《迁移托管智能体》。
• 把权限 Microsoft.CognitiveServices/accounts/AIServices/agents/endpoints/UserIdentityImpersonation/action 赋给你的中间层服务身份。内置角色不含此权限；通过自定义角色授予它，见《委托最终用户身份》。没有它，x-ms-user-identity 标头会以 403 被拒绝。
• 中间层使用 Azure AI Projects 客户端库，容器使用 Azure AI AgentServer 软件开发工具包（SDK）（Python 为 azure-ai-agentserver-core 2.0.0b7 及以上，.NET 为 Azure.AI.AgentServer.Core 1.0.0-beta.26 及以上）。
• 一个已部署的智能体，用来做测试。本地运行不强制隔离。

在共享会话中隔离两个用户

从核心行为开始：两个用户——称他们为 Alice 和 Bob，也就是示例里被代为行事的用户——可以共用一个 agent_session_id，平台仍然让每个用户的对话保持私有。你的中间层在每次调用时用 x-ms-user-identity 标头（委托）标明被代为行事的用户。要继续某个用户自己的对话，它把该用户的上一条响应作为 previous_response_id 传入。

要从 Azure Developer CLI 发送同一个身份标头，使用 azd ai agent invoke --user-identity。示例，以及如何选择该用户的会话和对话状态，见《传入最终用户身份》。

示例里那个最小的调用方 invoke_previous_response_isolation.py 发送的正是这些内容，用的是 SDK 里绑定到智能体的响应（Responses）客户端：

# Agent-bound Responses client from the Foundry SDK.
responses_client = project_client.get_openai_client(agent_name=agent_name).responses

# Target the shared session with agent_session_id, and identify the acted-for
# user with x-ms-user-identity (delegation). Pass previous_response_id to
# continue this user's own chain. Don't send x-agent-user-id; Foundry sets the
# container-side request context after it resolves the user.
kwargs = {
    "input": user_message,
    "stream": False,
    "store": True,
    "extra_body": {"agent_session_id": session_id},
    "extra_headers": {"x-ms-user-identity": user_id},
}
if previous_response_id:
    kwargs["previous_response_id"] = previous_response_id

response = responses_client.create(**kwargs)

平台把每条响应链绑到创建它的用户。如果 Bob 待在同一会话里，却发送 Alice 的 previous_response_id，这次调用会失败——Bob 不能继续 Alice 的对话。这个保证不需要你在容器里再写额外的隔离代码。

用会话池扩展到很多用户

在一个会话里隔离两个用户，是基本构件。要服务很多用户，把他们分到一组有上限的会话里，而不是每个用户开一个会话。

每个会话在主动处理一轮时，都会计入区域并发会话限制，所以每个用户一个会话扩不上去。因为用户在两轮之间会阅读、思考和输入，峰值并发请求通常只占用户总数的一小部分。按这个峰值给池定大小，再把每个用户映射到池中的一个会话，并在每次调用时传入该用户的身份，做法与上一节完全相同。

决定如何把用户映射到会话。常见策略包括：

• 粘性、最少负载。 回来的用户复用自己的会话；新用户进入负载最低的会话。此策略把负载摊匀，并把一个用户的各轮保持在一起。当会话达到每用户上限时，扩大这个池。
• 基于散列。 用 hash(user_id) % pool_size 分配会话。此策略简单且无状态，但负载可能不均，调整池的大小会把用户重新打乱。
• 轮询。 把请求均匀分到整个池。此策略简单，但一个用户的各轮可能落到不同会话。
• 基于分组。 按租户、团队或区域来路由，让相关用户共用会话。当一组用户共享上下文时，此策略有用。

示例里的调用方 invoke_session_pool.py 用两种策略实现由调用方自己拥有的分配：sticky-fill 和 round-robin。回来的用户始终保留自己的会话；新用户按所选策略安置。粘性填充这条路径会去填还有空位的会话，只有当每个会话都满了，才新开一个：

def get_session_for_user(self, user_id: str) -> str:
    if user_id in self.user_to_session:
        return self.user_to_session[user_id]      # returning user is sticky
    session_id = self._next_fill_session()        # new user: place by strategy
    self.user_to_session[user_id] = session_id
    self.session_user_counts[session_id] += 1
    return session_id

def _next_fill_session(self) -> str:
    # Reuse a session with capacity; open a new one only when all are full.
    session_id = next(
        (s for s, count in self.session_user_counts.items()
         if count < self.max_users_per_session),
        None,
    )
    if session_id is None:
        session_id = self._session_name(len(self.session_user_counts))
        self.session_user_counts[session_id] = 0
    return session_id

把返回的会话标识送进前面展示的同一个委托调用：它成为 extra_body 里的 agent_session_id，而 x-ms-user-identity 仍然是按用户区分的标识。

在容器里处理这次请求

在协议 2.0.0 上，平台解析被代为行事的用户，并通过 get_request_context() 把它交给你的处理函数。校验这个上下文（它缺失时按失败即关闭（fail closed）处理，例如本地运行），然后让平台用 context.get_history() 返回按用户划分的历史。示例里的处理函数 main.py 自己不保存对话状态：

from azure.ai.agentserver.core import get_request_context

@app.response_handler
async def handler(request, context, _cancellation_signal):
    ctx = get_request_context()
    if not (ctx.user_id and ctx.call_id):
        # Hosted protocol 2.0.0 populates this context; off-platform it's absent.
        raise ValueError("A user context is required on protocol 2.0.0.")

    user_input = await context.get_input_text() or "Hello!"
    history = await context.get_history()       # platform-authorized for this user
    input_items = _build_input(user_input, history)

    response = _responses_client.create(model=_model, input=input_items, store=False)
    return TextResponse(context, request, text=response.output_text)

因为平台按每次请求授权 context.get_history()，共享会话里的用户永远拿不到另一个用户的对话历史。

为容器自己存储的按用户数据做分区

平台为你隔离对话历史。如果容器另外还存储自己的数据——文件、数据库行或缓存——这些数据不会自动分区。同时用会话标识和用户标识做键，这样同一会话里的两个用户就看不到彼此的数据：

partition = (agent_session_id, user_id)

警告

当用户共用一个会话时，平台不会为容器自己存储的数据做分区。如果容器只按会话标识给这些数据做键，池里的每个用户都会看到同一份数据。分区键里始终要带上用户标识。

从每次请求的平台上下文读取用户标识。

Python

from azure.ai.agentserver.core import get_request_context

def partition_key() -> tuple[str, str]:
    ctx = get_request_context()
    if not ctx or not ctx.user_id:
        raise PermissionError("A user context is required on protocol 2.0.0.")
    return (ctx.session_id, ctx.user_id)   # key all user-owned data by this

平台也会把该用户作为请求标头 x-agent-user-id 注入。如果你的运行时不用 SDK 上下文，就直接读这个标头。

平台在协议 2.0.0 上填充 get_request_context().user_id。当不止一个用户可能进入这个会话时，绝不要只用会话标识作为用户自有数据的键。

C#

using Azure.AI.AgentServer.Core;

static (string SessionId, string UserId) PartitionKey()
{
    var ctx = FoundryAgentRequestContext.Current;
    if (string.IsNullOrEmpty(ctx.UserId))
    {
        throw new UnauthorizedAccessException(
            "A user context is required on protocol 2.0.0.");
    }
    return (ctx.SessionId!, ctx.UserId!);   // key all user-owned data by this
}

平台在协议 2.0.0 上填充 FoundryAgentRequestContext.Current.UserId。当不止一个用户可能进入这个会话时，绝不要只用会话标识作为用户自有数据的键。

关于按会话存储、可以接着往下做的实例，见记笔记智能体示例。它在 $HOME 下按每个会话键一个文件。对于共享会话，用请求上下文里的用户标识把这个键再加长，让每个用户得到自己的分区。

验证隔离

用示例的 A-A-B 测试 invoke_previous_response_isolation.py 来确认这个保证。用两个不同的用户（示例默认为 Alice 和 Bob）对你已部署的智能体运行它：

1. 作为 Alice，在一个共享会话里创建一条响应，并捕获它的 id。
2. 作为 Alice，在同一会话里创建第二条响应，把 previous_response_id 设为第一条响应的 id，并捕获它的 id。
3. 作为 Bob，在同一会话里发送请求，把 previous_response_id 设为 Alice 的第二条响应。这次调用失败——Bob 不能接着 Alice 的链往下走。

使用两个不同的 Entra 用户或对象标识。两个最终解析成同一身份的标签，不是一次有效的跨用户测试。

在协议 2.0.0 的路径上发送旧的隔离标头会返回错误，因为那种模型已经被平台用户上下文取代。

PRIMARY_SOURCE: https://learn.microsoft.com/en-us/azure/foundry/agents/how-to/multiplex-session-users