Multiplexar varios usuarios en una sesión de agente alojado

De forma predeterminada, cada llamador obtiene su propia sesión de agente hospedada, como se describe en Aislar sesiones de agente hospedadas por usuario. Las aplicaciones que sirven a muchos usuarios: un bot de Teams, una puerta de enlace de ISV o una plataforma de soporte técnico al cliente, no necesitan una sesión por usuario. En su lugar, un servicio de nivel intermedio asigna muchos usuarios a un grupo limitado de sesiones compartidas e identifica a cada usuario en cada llamada.

En este artículo se muestra cómo agrupar sesiones entre usuarios desde el nivel intermedio y mantener los datos de cada usuario aislados dentro de una sesión compartida.

La plataforma aísla para usted el estado de la conversación, incluso cuando los usuarios comparten una sesión: una cadena de respuestas que crea un usuario no puede ser continuada por otro usuario a través de previous_response_id, y context.get_history() devuelve solo el historial que el usuario de la solicitud actual está autorizado a ver. Usted controla dos cosas: la asociación entre usuario y sesión en su capa intermedia y la partición de cualquier dato que su contenedor almacene por sí mismo (archivos, filas o caché), más allá de ese estado de conversación gestionado por la plataforma.

Un ejemplo completo y ejecutable de multiplexación de sesiones muestra ambas partes —el grupo de sesiones de nivel intermedio y el controlador del contenedor—, y este artículo enlaza sus archivos a medida que avanza.

Prerrequisitos

• Agente hospedado que usa la versión 2.0.0 del protocolo de contenedor. Para actualizar, consulte Migración de agentes hospedados.
• El permiso Microsoft.CognitiveServices/accounts/AIServices/agents/endpoints/UserIdentityImpersonation/action asignado a la identidad de tu servicio de capa intermedia. Este permiso no se incluye en roles integrados; concédalo a través de un rol personalizado; consulte Delegación de la identidad del usuario final. Sin él, el encabezado x-ms-user-identity se rechaza con un 403.
• La biblioteca cliente de Azure AI Projects para el nivel intermedio y el SDK Azure AI AgentServer para el contenedor (azure-ai-agentserver-core 2.0.0b7+ para Python o Azure.AI.AgentServer.Core 1.0.0-beta.26+ para .NET).
• Un agente desplegado con el que realizar pruebas. El aislamiento no se aplica a las ejecuciones locales.

Aislar dos usuarios en una sesión compartida

Empiece por el comportamiento principal: dos usuarios —llamémoslos Alice y Bob, los usuarios representados en el ejemplo— pueden compartir un mismo agent_session_id, y la plataforma sigue manteniendo privadas las conversaciones de cada usuario. Tu capa intermedia identifica al usuario en cuyo nombre se actúa en cada llamada con el encabezado x-ms-user-identity (delegación). Para continuar la conversación de un usuario, pasa la respuesta anterior del usuario como previous_response_id.

El llamante invoke_previous_response_isolation.py mínimo del ejemplo envía exactamente eso, utilizando el cliente de respuestas vinculado al agente del SDK:

# 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)

La plataforma vincula cada cadena de respuesta al usuario que lo creó. Si Bob envía el previous_response_id de Alice mientras se encuentra en la misma sesión, la llamada genera un error: Bob no puede continuar la conversación de Alice. Esa garantía se mantiene sin necesidad de código de aislamiento adicional en tu contenedor.

Escalar a muchos usuarios con un grupo de sesiones

Aislar a dos usuarios dentro de una misma sesión es el elemento básico. Para atender a muchos usuarios, agrúpelos en un conjunto limitado de sesiones en lugar de abrir una sesión por usuario.

Cada sesión cuenta para los límites regionales de sesiones simultáneas mientras procesa activamente un turno, por lo que una sesión por usuario no es escalable. Dado que los usuarios leen, piensan y escriben entre turnos, las solicitudes simultáneas máximas suelen ser una pequeña fracción del recuento total de usuarios. Dimensiona el grupo para ese pico, luego asigna a cada usuario una sesión del mismo y pasa la identidad de ese usuario en cada llamada, exactamente como en la sección anterior.

Decida cómo asignar usuarios a sesiones. Entre las estrategias comunes se incluyen:

• Fija, con menor carga. Un usuario que devuelve reutiliza su sesión; los nuevos usuarios van a la sesión menos cargada. Esta estrategia distribuye la carga de manera uniforme y mantiene juntos los turnos de un usuario. Aumente el grupo cuando las sesiones alcancen un límite por usuario.
• Basado en hash. Asigne una sesión con hash(user_id) % pool_size. Esta estrategia es sencilla y no requiere estado, pero la carga puede ser desigual y el redimensionamiento del grupo reorganiza a los usuarios.
• Rotación por turnos. Distribuya las solicitudes uniformemente en el grupo. Esta estrategia es sencilla, pero los turnos de un usuario pueden recaer en diferentes sesiones.
• Basado en grupos. Dirige por tenant, equipo o región para que los usuarios relacionados compartan sesiones. Esta estrategia es útil cuando los usuarios de un grupo comparten contexto.

El invoke_session_pool.py llamante del ejemplo implementa la asignación administrada por el llamante con dos estrategias, sticky-fill y round-robin. Un usuario que devuelve siempre mantiene su sesión; la estrategia seleccionada coloca un nuevo usuario. La ruta sticky-fill rellena la sesión menos cargada y abre una nueva solo cuando todas las sesiones están al máximo de su capacidad:

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

Introduzca el identificador de sesión devuelto en la misma llamada delegada mostrada anteriormente: pasa a ser agent_session_id en extra_body, y x-ms-user-identity sigue siendo el identificador por usuario.

Gestiona la solicitud en tu contenedor

En el protocolo 2.0.0, la plataforma resuelve el usuario en nombre del cual se actúa y lo expone a tu controlador a través de get_request_context(). Valida ese contexto (cierra con error si falta, como en ejecuciones locales) y, a continuación, deja que la plataforma devuelva el historial por usuario con context.get_history(). El controlador main.py del ejemplo no mantiene ningún estado de conversación propio:

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)

Dado que la plataforma autoriza context.get_history() por solicitud, un usuario de una sesión compartida nunca recibe el historial de conversaciones de otro usuario.

Divide los datos de cada usuario que almacena tu contenedor

La plataforma mantiene aislado tu historial de conversaciones. Si el contenedor también almacena sus propios datos ( archivos, filas de base de datos o una memoria caché), esos datos no se particionan automáticamente. Haz que dependa tanto del ID de sesión como del ID de usuario para que dos usuarios en la misma sesión no puedan ver los datos del otro:

partition = (agent_session_id, user_id)

Leer el ID de usuario del contexto de plataforma para cada solicitud:

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

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
}

Para ver un ejemplo práctico de almacenamiento por sesión sobre el que basarse, consulte el ejemplo de agente para tomar notas. Almacena un archivo por sesión bajo $HOME. Para una sesión compartida, extienda esa clave con el identificador de usuario desde el contexto de solicitud para que cada usuario obtenga su propia partición.

Comprobación del aislamiento

Confirme la garantía con la prueba A-A-B de la muestra, invoke_previous_response_isolation.py. Ejecútelo con su agente desplegado usando dos usuarios distintos (el ejemplo usa de forma predeterminada a Alice y Bob):

1. Como Alice, crea una respuesta en una sesión compartida y captura su id.
2. Como Alice, crea una segunda respuesta en la misma sesión con previous_response_id establecido en el id de la primera respuesta, y captura su id.
3. Como Bob, en la misma sesión, envía una solicitud con previous_response_id establecido en la segunda respuesta de Alice. La llamada genera un error: Bob no puede continuar la cadena de Alice.

Use dos identificadores de objeto o usuarios de Entra diferentes. Dos etiquetas que corresponden a la misma identidad no constituyen una prueba válida entre usuarios.

El envío de un encabezado heredado de aislamiento en una ruta del protocolo 2.0.0 genera un error, ya que ese modelo ha sido sustituido por el contexto de usuario de la plataforma.

Contenido relacionado

• Aísle las sesiones del agente hospedado por usuario para el modelo de aislamiento predeterminado por llamador.
• Ejemplo de multiplexación de sesiones para el grupo completo de sesiones del nivel intermedio, el controlador de contenedores y la prueba de aislamiento.
• Ejemplo de agente de toma de notas para un contenedor que conserva los datos propiedad del usuario por sesión (Python y C#).
• Cuotas y límites para el servicio Foundry Agent para los límites regionales de sesiones simultáneas.
• Migrar agentes hospedados para migrar un contenedor al protocolo 2.0.0.
• Contrato de tiempo de ejecución del agente hospedado para los encabezados de la plataforma y las variables de entorno que recibe un contenedor.