Multiplexe vários usuários em uma sessão de agente hospedada

Por padrão, cada chamador obtém sua própria sessão de agente hospedado, conforme descrito em sessões de agente hospedado isolado por usuário. Aplicativos que atendem a muitos usuários - um bot do Teams, um gateway ISV ou uma plataforma de suporte ao cliente - não precisam de uma sessão por usuário. Em vez disso, um serviço de camada intermediária mapeia muitos usuários para um pool limitado de sessões compartilhadas e identifica cada usuário em cada chamada.

Este artigo mostra como agrupar sessões entre usuários da camada intermediária, mantendo os dados de cada usuário isolados dentro de uma sessão compartilhada.

A plataforma isola o estado da conversa para você, mesmo quando os usuários compartilham uma sessão: uma cadeia de respostas criada por um usuário não pode ser continuada por outro usuário através de previous_response_id, e context.get_history() retorna apenas o histórico que o usuário da solicitação atual está autorizado a ver. Você possui duas coisas: o mapeamento de usuário para sessão na camada intermediária e o particionamento de todos os dados que seu contêiner armazena sozinho (arquivos, linhas ou cache) além desse estado de conversa gerenciado pela plataforma.

Um exemplo completo e executável de multiplexação de sessão demonstra ambos os lados — o pool de sessões da camada intermediária e o manipulador de contêiner — e este artigo apresenta links para seus arquivos ao longo do texto.

Pré-requisitos

  • Um agente hospedado que usa o protocolo de contêiner versão 2.0.0. Para atualizar, consulte Migrar agentes hospedados.
  • A permissão Microsoft.CognitiveServices/accounts/AIServices/agents/endpoints/UserIdentityImpersonation/action atribuída à identidade do serviço de camada intermediária. Essa permissão não está incluída em funções predefinidas; conceda-a por meio de uma função personalizada — consulte Delegar a identidade do usuário final. Sem ele, o x-ms-user-identity cabeçalho é rejeitado com 403.
  • A biblioteca de cliente do Azure AI Projects para a camada intermediária e o SDK do Azure AI AgentServer para o contêiner (azure-ai-agentserver-core 2.0.0b7+ para Python, ou Azure.AI.AgentServer.Core 1.0.0-beta.26+ para .NET).
  • Um agente implantado para testar. O isolamento não é imposto para execuções locais.

Isolar dois usuários em uma sessão compartilhada

Comece com o comportamento básico: dois usuários — vamos chamá-los de Alice e Bob, os usuários representados no exemplo — podem compartilhar um agent_session_id, e a plataforma ainda mantém privadas as conversas de cada usuário. Sua camada intermediária identifica o usuário representado em cada chamada com o cabeçalho x-ms-user-identity (delegação). Para continuar a conversa de um usuário, ele passa a resposta anterior desse usuário como previous_response_id.

A implementação mais simples de chamada invoke_previous_response_isolation.py no código de exemplo envia exatamente isso, usando o cliente Responses do SDK vinculado ao agente:

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

A plataforma vincula cada cadeia de resposta ao usuário que a criou. Se Bob enviar o previous_response_id da Alice enquanto estiver na mesma sessão, a chamada falha — Bob não consegue continuar a conversa da Alice. Essa garantia é mantida sem nenhum código de isolamento extra em seu contêiner.

Escalar para muitos usuários com um pool de sessões

Isolar dois usuários em uma sessão é o bloco de construção. Para atender a muitos usuários, agrupe-os em um conjunto limitado de sessões em vez de abrir uma sessão por usuário.

Cada sessão é contabilizada nos limites regionais de sessões simultâneas enquanto processa ativamente uma interação, portanto uma sessão por usuário não escala. Como os usuários leem, pensam e digitam entre turnos, suas solicitações simultâneas de pico normalmente são uma pequena fração da sua contagem total de usuários. Dimensione um pool para esse pico, mapeie cada usuário para uma sessão nele e passe a identidade desse usuário em cada chamada, exatamente como na seção anterior.

Decida como mapear usuários para sessões. As estratégias comuns incluem:

  • Persistente, com menor carga. Um usuário que retorna reutiliza sua sessão; novos usuários vão para a sessão menos carregada. Essa estratégia distribui a carga de modo uniforme e mantém juntos os turnos de um usuário. Aumente o pool quando as sessões atingirem um limite por usuário.
  • Baseado em hash. Atribua uma sessão com hash(user_id) % pool_size. Essa estratégia é simples e sem estado, mas a carga pode ser desigual e o redimensionamento do pool redistribui usuários.
  • Round-robin. Distribua solicitações uniformemente pelo pool. Essa estratégia é simples, mas os turnos de um usuário podem cair em sessões diferentes.
  • Baseado em grupos. Rotear por locatário, equipe ou região para que os usuários relacionados compartilhem sessões. Essa estratégia é útil quando os usuários de um grupo compartilham contexto.

O chamador invoke_session_pool.py no exemplo implementa a atribuição pertencente ao chamador com duas estratégias, sticky-fill e round-robin. Um usuário que retorna sempre mantém sua sessão; um novo usuário é colocado pela estratégia selecionada. A estratégia de preenchimento fixo preenche primeiro a sessão com menor carga e só abre uma nova quando todas as sessões atingem a capacidade máxima:

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

Alimente a ID da sessão retornada para a mesma chamada delegada mostrada anteriormente: ela se torna agent_session_idextra_bodye x-ms-user-identity permanece o identificador por usuário.

Processar a solicitação no seu contêiner

No protocolo 2.0.0, a plataforma resolve o usuário representado e o disponibiliza ao seu manipulador por meio de get_request_context(). Valide esse contexto (fail closed quando ele estiver ausente, como em execuções locais), em seguida, permita que a plataforma retorne o histórico por usuário com context.get_history(). O manipulador do exemplo main.py não mantém seu próprio estado da conversa:

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)

Como a plataforma autoriza context.get_history() por solicitação, um usuário em uma sessão compartilhada nunca recebe o histórico de conversas de outro usuário.

Particionar dados por usuário que seu contêiner armazena

A plataforma isola o histórico de conversas para você. Se o contêiner também armazenar seus próprios dados - arquivos, linhas de banco de dados ou um cache - esses dados não serão particionados automaticamente. Use o ID da sessão e o ID do usuário como chave para que dois usuários na mesma sessão não consigam ver os dados um do outro:

partition = (agent_session_id, user_id)

Aviso

Quando os usuários compartilham uma sessão, a plataforma não particiona os dados que seu contêiner armazena sozinho. Se o seu contêiner usar apenas o ID da sessão como chave para esses dados, todos os usuários do pool verão os mesmos dados. Inclua sempre a ID do usuário na chave de partição.

Leia o ID do usuário do contexto de plataforma de cada solicitação:

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

A plataforma também injeta o usuário como o cabeçalho de solicitação x-agent-user-id. Se o runtime não usar o contexto do SDK, leia esse cabeçalho diretamente.

A plataforma preenche get_request_context().user_id no protocolo 2.0.0. Nunca use apenas a ID da sessão para dados pertencentes ao usuário quando mais de um usuário puder entrar na sessão.

Para ver um exemplo prático de armazenamento por sessão para usar como base, consulte o exemplo do agente de anotações. Ele usa como chave um arquivo por sessão em $HOME. Para uma sessão compartilhada, estenda essa chave com a ID do usuário do contexto de solicitação para que cada usuário obtenha sua própria partição.

Verificar isolamento

Confirme a garantia com o teste A-A-B da amostra, invoke_previous_response_isolation.py. Execute-o em seu agente implantado com dois usuários distintos (o exemplo usa como padrão Alice e Bob):

  1. Como Alice, crie uma resposta em uma sessão compartilhada e capture sua id.
  2. Como Alice, crie uma segunda resposta na mesma sessão, com previous_response_id definido como o id da primeira resposta, e capture seu id.
  3. Como Bob, na mesma sessão, envie uma solicitação com previous_response_id definido como a segunda resposta de Alice. A chamada falha — Bob não consegue continuar a sequência de Alice.

Use dois usuários ou IDs de objeto do Entra diferentes. Dois rótulos que apontam para a mesma identidade não constituem um teste válido entre usuários diferentes.

Enviar um cabeçalho de isolamento herdado em um caminho do protocolo 2.0.0 retorna um erro, pois esse modelo é substituído pelo contexto do usuário da plataforma.

Conteúdo relacionado

  • Last updated on