Gerenciar o estado de agentes de longa duração (versão prévia)

Um agente hospedado de longa duração recupera-se de falhas apenas se o seu progresso for persistente. Este artigo mostra como manter marcadores de progresso e pontos de verificação no repositório de estado do Foundry e como fazer backup de um ponto de verificação da estrutura para que a recuperação em nível de estrutura sobreviva às reinicializações.

Note

Agentes de longa duração estão em versão preliminar. APIs e versões de pacote estão sujeitas a alterações.

Armazenar progresso e pontos de verificação

FoundryStateStore é um armazenamento de chave-valor durável, com persistência no servidor, para armazenar o estado que deve sobreviver a falhas e ao despejo por inatividade. Mantenha marcadores de progresso e pontos de verificação em itens separados para que uma tentativa recuperada possa identificar a próxima etapa e os resultados já gravados.

Item O que ele contém Padrão de escrita
Marcador de progresso A etapa seguinte do fluxo de trabalho ou outro pequeno ponto de retomada. Atualize uma chave estável com a ETag da última leitura.
Checkpoint Um resultado concluído de uma etapa, um artefato gerado ou um estado serializado do framework. Crie um item com uma chave determinística somente de anexação.

Um armazenamento está associado a um nome escolhido pelo chamador. Inclua o fluxo de trabalho, a sessão, a thread ou o escopo da execução nesse nome. Para ver a visão geral completa do armazenamento, consulte Armazenamento de estado durável para agentes hospedados.

from azure.ai.agentserver.core.storage import FoundryStateStore

async def save_step(task_id: str, step: int, result: dict) -> None:
    store = await FoundryStateStore.get_or_create(f"workflows/{task_id}")
    async with store:
        progress = await store.get_item("progress")
        if progress and int(progress.value["workflow_step"]) > step:
            return

        checkpoint_key = f"checkpoints/{step}"
        checkpoint = await store.get_item(checkpoint_key)
        if checkpoint is None:
            await store.create_item(f"checkpoints/{step}", result)
        elif checkpoint.value != result:
            raise RuntimeError("The checkpoint contains a different result.")

        next_progress = {"workflow_step": step + 1}
        if progress is None:
            await store.create_item("progress", next_progress)
        else:
            await store.set_item(
                "progress", next_progress, if_match=progress.etag
            )

Escreva o ponto de verificação antes de avançar o marcador de progresso. Se o processo falhar entre essas gravações, a chave de ponto de verificação determinística permitirá que a tentativa recuperada detecte a gravação concluída. Em uma 409 Conflict ou 412 Precondition Failed resposta, leia novamente os itens e reconcilie o ponto de verificação existente antes de tentar novamente.

Principais comportamentos:

  • get_or_create() busca ou cria o repositório em uma chamada. Aplica-se user_isolation e item_ttl_seconds somente na primeira criação.
  • Nome da loja = escopo. Os nomes podem conter /; use-o como separador de hierarquia e escolha um esquema estável antecipadamente.
  • Concorrência otimista. Use if_match=item.etag para itens mutáveis, como marcadores de progresso. Use create_item() para pontos de verificação somente de anexação, para que uma gravação concorrente não substitua um valor existente.
  • Limites. Um valor de item pode conter até 1 MB de JSON serializado. O nome da loja pode conter de 1 a 128 caracteres e um item pode ter até 16 tags.

Implementar um mecanismo de checkpoint para o framework

Aponte um checkpointer do LangGraph ou do Microsoft Agent Framework (MAF) para FoundryStateStore e a recuperação do próprio framework se torna persistente mesmo após falhas, sem código de recuperação personalizado.

Conceito de estrutura FoundryStateStore
Tópico / escopo Nome da loja (codifique o ID nele)
ID do ponto de verificação Chave do item
Ponto de verificação serializado Valor do item (JSON dict)
mais recente / histórico / filtragem Etiquetas + list_keys(order="desc")
Segurança por usuário user_isolation=True

Os pontos de verificação operam apenas por adição: cada salvamento utiliza um novo ID, portanto, não há contenção de gravação e nunca é necessário if_match no caminho do ponto de verificação.

# LangGraph: one thread = one store
async def _store(thread_id: str) -> FoundryStateStore:
    return await FoundryStateStore.get_or_create(
        f"langGraphCheckpoints/{thread_id}", user_isolation=True
    )

Warning

Para o adaptador MAF, sempre defina user_isolation=True. O único agrupamento do MAF é workflow_name - um nome de definição compartilhado entre os usuários - portanto, sem isolamento do usuário, get_latest / list_checkpoints retornaria os pontos de verificação de outros chamadores.

Mantenha os campos de entrada pequenos

As entradas da tarefa são persistidas antes da execução do manipulador (é nisso que a recuperação se baseia), portanto mantenha-as pequenas — o limite por entrada individual é de cerca de 10 MiB após a serialização em JSON, e entradas maiores geram InputTooLarge antes de qualquer chamada de rede. Externalize grandes cargas de dados para o armazenamento de blobs e passe uma referência.

Conteúdo relacionado

  • Última atualização em