在單一裝載 Agent 工作階段中的多工多位使用者

預設情況下，每位呼叫者擁有自己的託管代理會話，詳見《 隔離每位使用者的託管代理會話》。 服務多用戶的應用程式——如 Teams 機器人、ISV 閘道或客戶支援平台——不需要每位使用者只設一個會話。 取而代之的是，中階服務將多個使用者映射到一個有界的共享會話池，並在每次通話中識別每位使用者。

這篇文章將教你如何在中階層的使用者間合併會話，同時將每個使用者的資料隔離在共享會話中。

平台會為你隔離對話狀態，即使使用者共用工作階段亦然：某位使用者建立的回應鏈，另一位使用者也無法透過 previous_response_id 接續，而 context.get_history() 只會傳回目前請求的使用者獲授權查看的歷史記錄。 您負責兩件事：在中介層中維護使用者到工作階段的對應，以及除了平台管理的對話狀態，還會對容器本身儲存的任何資料 (檔案、資料列或快取) 進行分區。

完整且可執行的工作階段多工範例展示了中介層工作階段集區與容器處理常式的兩端，且本文也會在過程中連結其檔案。

先決條件

  • 一個使用容器協定版本 2.0.0 的託管代理。 要升級，請參見遷移託管代理。
  • Microsoft.CognitiveServices/accounts/AIServices/agents/endpoints/UserIdentityImpersonation/action 權限已指派給您的中間層服務身分識別。 此權限不包含在內建角色中;透過自訂角色授予——參見 「委派終端使用者身份」。 若缺少它，x-ms-user-identity 標頭會遭到拒絕，並傳回 403。
  • 中階用 Azure AI Projects 客戶端函式庫，容器用 Azure AI AgentServer SDK（azure-ai-agentserver-corePython 用 2.0.0b7+，Azure.AI.AgentServer.Core.NET 用 1.0.0-beta.26+）。
  • 一個用於測試 Agent 的已部署 Agent。 本機執行未強制隔離。

在共享會話中隔離兩名使用者

從核心行為開始：兩個使用者——稱他們為 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 中的 Agent 繫結回應用戶端，傳送內容：

# 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

將傳回的工作階段 ID 輸入先前所示的委派呼叫：它會變成 agent_session_id 的 extra_body，並x-ms-user-identity維持每位使用者的識別碼。

在容器中處理要求

在協定 2.0.0 中，平台會解析代表的使用者，並透過 get_request_context()，將其提供給處理常式。 驗證前後關聯 (若缺失，以封閉方式失敗，例如在本機執行時)，然後讓平台透過 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() 請求進行授權，因此共用工作階段中的使用者絕不會取得其他使用者的對話紀錄。

將容器儲存的各使用者資料加以劃分

平台會幫你隔離對話紀錄。 如果你的容器 也 儲存自己的資料——檔案、資料庫列或快取——那些資料不會自動分割。 同時用會話 ID 和使用者 ID 來標記，這樣同一會話中的兩個使用者就無法看到彼此的資料：

partition = (agent_session_id, user_id)

警告

當使用者共享會話時，平台不會將容器本身儲存的資料分割。 如果你的容器只用會話 ID 來鍵入該資料，池中每個使用者都會看到相同的資料。 分割鍵中一定要包含使用者 ID。

從每個請求平台的上下文中讀取使用者 ID：

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。 當多個使用者可以進入會話時，切勿僅用會話 ID 來管理使用者擁有的資料。

如需做為後續建置基礎的每工作階段儲存體可行範例，請參閱筆記記錄 Agent 範例。 它會在 $HOME 下，將每個工作階段設定為一個檔案。 對於共享會話，從請求上下文中延伸該金鑰並加入使用者 ID，讓每個使用者擁有自己的分割區。

驗證隔離狀態

用樣本的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 使用者或物件 ID。 兩個標籤解析為相同身份，並不構成有效的跨使用者測試。

在協定 2.0.0 路徑上傳送舊有隔離標頭會回傳錯誤，因為該模型已被平台使用者上下文取代。

相關內容

  • 上次更新時間：