Tutorial: adicionar sessões e memória a um agente existente

info

Beta

Esse recurso está na versão Beta.

Se você já criou um agente em Python, você pode dar a ele um histórico de conversa durável e memória de longo prazo no Databricks sem reescrevê-lo. Neste tutorial, você conectará o agente a um armazenamento de sessão que persiste cada conversa, um armazenamento de memória que mantém fatos duráveis ao longo das conversas e um pipeline de memória que destila as conversas armazenadas na memória. Os passos abrangem agentes criados com o OpenAI Agents SDK, LangGraph ou qualquer outro framework Python, onde quer que o agente seja executado.

dica

Se você ainda não tiver um agente, comece com o início rápido do Agent Bricks. A CLI do Agent Bricks cria um novo agente com sessões e memória já configuradas.

Neste exemplo, um agente de suporte deve retomar a conversa de um cliente de onde parou e lembrar as preferências do cliente em conversas posteriores e separadas.

Para saber como as sessões e a memória se relacionam, consulte Memória e sessões do agente.

Requisitos

Um agente existente escrito em Python 3.10 ou acima. Os passos específicos do framework cobrem o OpenAI Agents SDK e o LangGraph. Os passos para outros frameworks chamam o SDK AgentKit diretamente.
Autenticação Databricks. O SDK do AgentKit usa a autenticação default do SDK do Databricks para Python. Para desenvolvimento local, execução databricks auth login --host <workspace-url> ou defina as variáveis de ambiente DATABRICKS_HOST e DATABRICKS_TOKEN.
A CLI do Databricks, para verificar o status das execuções do pipeline de memória.
Para criar um memory pipeline, é necessária permissão para fazer query no modelo que o pipeline utiliza e no outro acesso listado nos requisitos do memory pipeline.

O passo 1: instalar o SDK do AgentKit e o CLI do Agent Bricks

O pacote databricks-agentbricks contém o seguinte:

O SDK do AgentKit (databricks_agentkit), o cliente Python para lojas de sessão e memória.
A CLI do Agent Bricks (agentbricks).
Adaptadores para o OpenAI Agents SDK e LangGraph que conectam a sessão e os armazenamentos de memória às interfaces de sessão e ferramentas existentes de cada framework. Os adaptadores usam os pacotes de framework que o seu agente já tem instalado.

Instale a versão 0.5.0 ou acima no ambiente do seu agente:

Bash
pip install --upgrade "databricks-agentbricks>=0.5.0"

O passo 2: Crie um armazenamento de sessão e um armazenamento de memória

Seu agente usa dois repositórios que o Databricks gerencia para você:

Um armazenamento de sessão mantém a história da conversa. Cada sessão nele é a transcrição de uma conversa, que o agente recarrega para continuar essa conversa de onde parou.
Um armazenamento de memória mantém fatos duráveis sobre cada usuário, como preferências e decisões anteriores. O agente a pesquisa para recordar o que aprendeu em conversas anteriores e separadas.

O store é o limite de segurança: qualquer principal com acesso a um store pode ler e gravar todas as suas entradas. Para isolar os tenants uns dos outros, crie um par separado de stores para cada tenant. Consulte Particionar e proteger a memória.

Crie cada armazenamento uma vez e reutilize-o para todas as conversas. Seu agente se refere a ambos os armazenamentos pelo nome.

AgentKit SDK
Agent Bricks CLI
REST API

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

client = AgentKitClient(WorkspaceClient())
client.session_stores.create("support-agent-sessions", description="Support agent conversations")
client.memory_stores.create("support-agent-memory", description="Support agent long-term memory")

Bash
agentbricks sessions stores create --name support-agent-sessions --description "Support agent conversations"
agentbricks memory stores create --display-name support-agent-memory --description "Support agent long-term memory"

Defina DATABRICKS_HOST como o hostname do seu workspace, depois execução o seguinte:

Bash
export DATABRICKS_TOKEN=$(databricks auth token --profile <profile> | jq -r .access_token)

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/session-stores?session_store_name=support-agent-sessions" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"description": "Support agent conversations"}'

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-stores" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"display_name": "support-agent-memory", "description": "Support agent long-term memory"}'

Os nomes de armazenamento devem ser exclusivos no workspace. O nome de um armazenamento de memória deve ter entre 3 e 56 caracteres, começar com uma letra minúscula, terminar com uma letra ou um número e conter apenas letras minúsculas, números e hifens. O nome do armazenamento de sessão deve ter entre 3 e 63 caracteres.

O passo 3: Escolher IDs de ator e de sessão

Cada chamada que seu agente faz para os repositórios carrega dois identificadores:

actor_id identifica de quem são as conversas e memórias. Ambos os stores o usam, e um pipeline de memória grava as memórias de cada ator sob o mesmo actor_id que as sessões desse ator, para que o agente recorde memórias destiladas apenas quando os dois corresponderem.
session_id identifica uma conversa, de modo que uma solicitação de acompanhamento a retoma.

Para este tutorial, use test-user como actor_id e conversation-1 como session_id. As funções handle_message nas etapas seguintes os utilizam como user_id e conversation_id.

Na produção, defina actor_id no código da aplicação a partir de contexto de solicitação confiável, como a identidade verificada do usuário conectado, e nunca deixe o modelo ou a mensagem de um usuário escolhê-lo. Para compartilhar memória entre uma equipe ou isolar tenants, consulte Particionar e proteger a memória. Para session_id, use o ID de conversa ou tópico existente do seu aplicativo ou gere um UUID quando uma conversa começar.

O passo 4: Persistir conversas no armazenamento de sessão

A história de conversas é o registro ordenado de uma conversa: as mensagens do usuário, as respostas do agente e quaisquer chamadas de ferramenta e resultados. Um armazenamento de sessão salva essa história para cada conversa, para que o agente possa recarregá-lo na próxima solicitação, mesmo após uma reinicialização. Substitua a história de conversas em processo do seu agente pelo armazenamento de sessão que você criou no Passo 2.

Os adaptadores de estrutura conectam-se à interface que cada estrutura já usa para estado de sessão, para que o modelo, as ferramentas e os prompts do seu agente permaneçam os mesmos.

OpenAI Agents SDK
LangGraph
Other frameworks

O OpenAI Agents SDK lê e grava a história da conversa por meio de um objeto Session. session_store() retorna um Session com suporte do armazenamento de sessão gerenciado. Passe para Runner.run:

Python
from agents import Agent, Runner
from databricks_agentkit.openai import session_store

SESSION_STORE = "support-agent-sessions"

# Your existing agent, unchanged.
agent = Agent(name="Support agent", instructions="You help customers with their Databricks workspaces.")

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(agent, message, session=session)
    return result.final_output

Na primeira mensagem de uma conversa, o adaptador cria a sessão. A cada execução, o SDK carrega a história da conversa da sessão e adiciona a nova mensagem do usuário, a saída do modelo e os itens da ferramenta.

O LangGraph persiste o estado do gráfico por meio de um checkpointer vinculado a um ID de thread. checkpointer() retorna um checkpointer com backup do armazenamento de sessões gerenciado e thread_config() mapeia uma conversa e seu usuário em um thread:

Python
from langchain.agents import create_agent
from databricks_agentkit.langgraph import checkpointer, thread_config

SESSION_STORE = "support-agent-sessions"

# Your existing model and tools, unchanged, with the managed checkpointer added.
agent = create_agent(model=model, tools=tools, checkpointer=checkpointer(store=SESSION_STORE))

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": message}]},
        config=thread_config(conversation_id, actor=user_id),
    )
    return result["messages"][-1].content

Se o seu agente já usa um checkpointer, substitua-o por checkpointer(). Se você criar um StateGraph diretamente, passe o checkpointer para compile(). O checkpointer armazena o estado completo do gráfico, incluindo interrupções de intervenção humana, de modo que uma execução pausada possa ser retomada após um reinício.

nota

O verificador armazena pontos de verificação de gráfico serializados, que um pipeline de memória não lê. Para usar um pipeline de memória com um agente LangGraph, também grave uma transcrição.

Para qualquer outra estrutura, ou para chamadas diretas de modelo, leia e anexe a história da conversa com o SDK do AgentKit. Armazene cada mensagem como {"type": "message", "role": "user" | "assistant", "content": "<text>"}. Um pipeline de memória destila mensagens de usuário e assistente nesta forma, e na forma de mensagem de Respostas da OpenAI com partes de conteúdo de texto. Ele ignora todos os outros itens, como chamadas de ferramenta e resultados de ferramenta.

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

client = AgentKitClient(WorkspaceClient())
session_store = client.session_stores.get("support-agent-sessions")

def start_conversation(user_id: str, conversation_id: str) -> None:
    session_store.add(actor_id=user_id, session_id=conversation_id)

def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    session = session_store.get(conversation_id)
    if session.actor_id != user_id:
        raise PermissionError("This conversation belongs to a different user.")

    # Rebuild context from the stored history, oldest first.
    history = [item.data for item in session.list_items(order_by="create_time asc")]
    user_item = {"type": "message", "role": "user", "content": message}
    reply = run_agent([*history, user_item])  # Your framework or model call.

    session.append_items([user_item, {"type": "message", "role": "assistant", "content": reply}])
    return reply

A pesquisa de uma sessão por session_id não está definida para um ator, portanto, verifique se a sessão pertence ao usuário que a solicita antes de retomá-la.

A alternância para o armazenamento de sessão não move as conversas que seu agente armazenou em outro lugar. Para manter uma conversa existente, acrescente seus itens a uma nova sessão.

O passo 5: dar memória ao agente

A história da sessão cobre uma conversa. Para usar o que o agente aprendeu em conversas anteriores, conecte o agente ao armazenamento de memória. Existem dois padrões e você pode combiná-los:

Ferramentas de memória: O modelo decide quando lembrar da memória e quando salvar um novo fato.
Recordar antes de cada execução: sua aplicação pesquisa na memória a mensagem recebida e adiciona os resultados ao contexto do agente, de modo que o agente veja a memória relevante sem chamar uma ferramenta.

Se um pipeline de memória gravar sua memória (veja o Passo 6), o agente só precisa recordar a memória. Você pode pular ferramentas que salvam memória.

OpenAI Agents SDK
LangGraph
Other frameworks

O exemplo a seguir usa ambos os padrões em uma cópia por solicitação do seu agente:

memory_tools() retorna uma ferramenta recall que pesquisa a memória do usuário e uma ferramenta remember que salva um fato. As ferramentas são vinculadas a um actor, portanto o modelo não pode ler ou gravar na memória de outro usuário.
recalled_facts() pesquisa a memória da mensagem recebida antes da execução e adiciona os resultados às instruções do agente.

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient
from databricks_agentkit.openai import memory_tools, session_store

MEMORY_STORE = "support-agent-memory"
memory_store = AgentKitClient(WorkspaceClient()).memory_stores.get(MEMORY_STORE)

def recalled_facts(user_id: str, message: str) -> str:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    return "\n".join(f"- {r.memory.content}" for r in results)

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    instructions = agent.instructions  # Your agent's existing instructions, as a string.
    if facts := recalled_facts(user_id, message):  # Recall before the run.
        instructions += f"\n\nWhat you know about this user:\n{facts}"
    user_agent = agent.clone(
        instructions=instructions,
        tools=[*agent.tools, *memory_tools(user_id, store=MEMORY_STORE)],  # Memory tools.
    )
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(user_agent, message, session=session)
    return result.final_output

Os fatos recuperados vão para as instruções, não para a entrada, portanto, a sessão não os armazena. Para usar apenas um padrão, descarte o outro: remova a linha memory_tools() ou remova a chamada recalled_facts(). Se você mantiver as ferramentas, atualize as instruções do seu agente para dizer a ele quando usá-las, por exemplo: "Use a lembrança para procurar as preferências do usuário antes de responder. Use lembre-se de salvar fatos duráveis que o usuário compartilha."

O exemplo a seguir usa ambos os padrões e cria o agente para cada solicitação. O checkpointer é criado uma vez e compartilhado:

memory_tools() retorna uma ferramenta recall que pesquisa a memória do usuário e uma ferramenta remember que salva um fato. As ferramentas são vinculadas a um actor, portanto o modelo não pode ler ou gravar na memória de outro usuário.
recalled_facts() pesquisa na memória a mensagem de entrada antes da execução e adiciona os resultados ao prompt do sistema.

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient
from langchain.agents import create_agent
from databricks_agentkit.langgraph import checkpointer, memory_tools, thread_config

MEMORY_STORE = "support-agent-memory"
memory_store = AgentKitClient(WorkspaceClient()).memory_stores.get(MEMORY_STORE)

def recalled_facts(user_id: str, message: str) -> str:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    return "\n".join(f"- {r.memory.content}" for r in results)

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    system_prompt = SYSTEM_PROMPT  # Tell the agent when to use recall and remember.
    if facts := recalled_facts(user_id, message):  # Recall before the run.
        system_prompt += f"\n\nWhat you know about this user:\n{facts}"
    agent = create_agent(
        model=model,
        tools=[*tools, *memory_tools(user_id, store=MEMORY_STORE)],  # Memory tools.
        system_prompt=system_prompt,
        checkpointer=checkpointer(store=SESSION_STORE),
    )
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": message}]},
        config=thread_config(conversation_id, actor=user_id),
    )
    return result["messages"][-1].content

O LangGraph adiciona o prompt do sistema a cada chamada de modelo sem salvá-lo no estado do gráfico, portanto, o checkpointer não armazena os fatos recuperados. Para usar apenas um padrão, descarte o outro: remova a linha memory_tools() ou remova a chamada recalled_facts().

Pesquise na memória a mensagem recebida antes de cada execução e passe os resultados para o agente como contexto. A pesquisa classifica as entradas por relevância de texto completo, portanto, uma query criada a partir das próprias palavras do usuário funciona bem.

Python
memory_store = client.memory_stores.get("support-agent-memory")

def recall(user_id: str, message: str) -> list[dict]:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    if not results:
        return []
    facts = "\n".join(f"- {r.memory.content}" for r in results)
    return [{"role": "system", "content": f"What you know about this user:\n{facts}"}]

Em handle_message do Passo 4, adicione o contexto recuperado à entrada do agente. Não o anexe à sessão:

Python
reply = run_agent([*recall(user_id, message), *history, user_item])

Para permitir que o agente salve a memória sozinho, encapsule as operações search e add do store como ferramentas para o seu framework. Consulte Dar ferramentas de memória ao seu agente.

O passo 6: Criar um memory pipeline

Um pipeline de memória lê as conversas em seu armazenamento de sessão e destila fatos duráveis, preferências e decisões de cada ator no armazenamento de memória. Ele também faz merge de entradas sobrepostas ao longo do tempo, para que a memória do seu agente permaneça consolidada sem código extra do agente.

Crie um pipeline que conecte os dois stores:
Bash
```
agentbricks memory pipeline create \
  --session-store support-agent-sessions \
  --memory-store support-agent-memory \
  --trigger scheduled \
  --model system.ai.claude-sonnet-5 \
  --instructions "Focus on durable customer preferences and account details. Ignore one-off troubleshooting steps."
```
- --trigger scheduled executa o pipeline automaticamente a cada 24 horas. Sem isso, o pipeline usa o default Trigger, manual, e é executado apenas quando você começa uma execução. Consulte Execução de um pipeline ao programar.
- --model define o modelo que destila as memórias. Você pode usar qualquer modelo para o qual tenha permissão de query. Se você omitir --model, o pipeline usará um modelo default escolhido pelo Databricks.
- --instructions é opcional e orienta o que o pipeline mantém. Consulte Orientar a destilação com instruções.
O comando retorna o nome do recurso do pipeline, na forma memory-pipelines/<pipeline-id>. Para cada opção, faça a execução de agentbricks memory pipeline create --help ou consulte a referência de comando da CLI do Agent Bricks.
O programador escolhe quando a primeira execução programada começa. Para destilar as conversas que já estão no armazenamento de sessão, começar uma execução agora:
Bash
```
agentbricks memory pipeline run <pipeline-id>
```
O comando retorna o nome do recurso da execução, na forma memory-pipelines/<pipeline-id>/runs/<run-id>.

Verifique o resultado da execução com a API REST. Defina DATABRICKS_HOST como seu hostname de workspace e execute o seguinte:

Bash
export DATABRICKS_TOKEN=$(databricks auth token --profile <profile> | jq -r .access_token)
curl "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>/runs/<run-id>" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

Quando o state da execução é SUCCEEDED, as memórias destiladas estão no armazenamento de memória.

As execuções de pipeline usam a identidade que criou o pipeline, não a identidade do agente. Um pipeline pode reescrever, merge ou excluir qualquer entrada em um actor_id que ele processa, incluindo entradas que seu agente salvou com ferramentas de memória. Para saber mais sobre a API REST, estados de execução e limitações, consulte Pipelines de memória do agente.

Registrar uma transcrição para agentes LangGraph

Um pipeline de memória destila apenas mensagens de usuário e assistente. O checkpointer do LangGraph armazena pontos de verificação serializados, portanto, um pipeline que lê o armazenamento de sessão do checkpointer não encontra mensagens para destilar. Para usar um pipeline com um agente LangGraph, também anexe as mensagens de cada turno a um armazenamento de sessão separado e crie o pipeline nesse armazenamento:

Python
transcripts = client.session_stores.create("support-agent-transcripts")  # Create once.

def start_conversation(user_id: str, conversation_id: str) -> None:
    transcripts.add(actor_id=user_id, session_id=conversation_id)

def record_turn(conversation_id: str, message: str, reply: str) -> None:
    transcripts.get(conversation_id).append_items(
        [
            {"type": "message", "role": "user", "content": message},
            {"type": "message", "role": "assistant", "content": reply},
        ]
    )

Chame start_conversation quando uma conversa começar, antes do seu primeiro ainvoke, e chame record_turn após cada ainvoke. Passe --session-store support-agent-transcripts ao criar o pipeline. Como alternativa, ignore o pipeline e deixe o agente salvar a memória com ferramentas de memória.

O passo 7: Conceder acesso ao seu agente implantado

Quando você execução o agente localmente, ele usa suas próprias credenciais. Um agente implantado geralmente é executado como um Service Principal, que precisa de acesso a todos os stores que ele lê ou grava. Conceda acesso com o SDK do AgentKit, passando a ID do aplicativo (cliente) da Service Principal:

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

w = WorkspaceClient()
client = AgentKitClient(w)

# For an agent deployed as a Databricks app, look up the app's service principal.
sp_client_id = w.apps.get("support-agent").service_principal_client_id

client.session_stores.get("support-agent-sessions").grant_permission(sp_client_id)
client.memory_stores.get("support-agent-memory").grant_permission(sp_client_id)

grant_permission concede acesso de gravação por default. Para conceder acesso somente de leitura, passe permission="READ". Se você criou um armazenamento de transcrições para um agente LangGraph, conceda acesso a ele também. O pipeline de memória não precisa de uma concessão para o service principal do seu agente, porque as execuções de pipeline usam a identidade que criou o pipeline.

O passo 8: Testar o agente de ponta a ponta

Chame handle_message com user_id="test-user" e conversation_id="conversation-1" e compartilhe uma preferência durável, como "Prefiro email ao telefone e estou no fuso horário do Pacífico." Se você seguiu os passos para outras estruturas, chame start_conversation("test-user", "conversation-1") primeiro. Envie uma mensagem de acompanhamento na mesma conversa e confirme se o agente mantém o contexto anterior.

Confirme se o armazenamento da sessão contém a conversa:

Bash
agentbricks sessions items list --store support-agent-sessions --session-id conversation-1

Começar uma execução de pipeline de memória e aguarde até que o state seja SUCCEEDED, conforme o o passo 6.

Confirme se o armazenamento de memória mantém a preferência para o usuário de teste:

Bash
agentbricks memory entries search --store support-agent-memory --actor-id test-user --query "contact preferences"

Comece uma nova conversa para test-user com conversation_id="conversation-2" e pergunte “Como você deve entrar em contato comigo?” O agente responde da memória.

Solução de problemas

PERMISSION_DENIED (HTTP 403) do agente implantado. O Service Principal do agente não pode acessar um armazenamento. Conceda acesso a ele, conforme o Passo 7.
Uma execução de pipeline foi bem-sucedida, mas não gravou nenhuma entrada. Confirme se o armazenamento de sessão contém mensagens de usuários e assistentes em um formato que o pipeline lê. Um pipeline ignora chamadas de ferramenta, resultados de ferramenta e checkpoints do LangGraph. Uma execução processa apenas as sessões de um ator desde a última execução bem-sucedida desse ator.
O agente não se lembra de memórias destiladas. Confirme se o agente pesquisa na memória com o mesmo actor_id que suas sessões usam.
A execução inicia com falha de ABORTED. Outra execução do pipeline está em andamento. Aguarde até que seja concluído e, em seguida, começar uma nova execução.

Limpar

Para excluir os recursos que este tutorial criou, execução o seguinte. Excluir uma loja exclui tudo o que há nela.

Bash
agentbricks memory pipeline delete <pipeline-id>
agentbricks sessions stores delete support-agent-sessions
agentbricks memory stores delete support-agent-memory

Requisitos​

O passo 1: instalar o SDK do AgentKit e o CLI do Agent Bricks​

O passo 2: Crie um armazenamento de sessão e um armazenamento de memória​

O passo 3: Escolher IDs de ator e de sessão​

O passo 4: Persistir conversas no armazenamento de sessão​

O passo 5: dar memória ao agente​

O passo 6: Criar um memory pipeline​

Registrar uma transcrição para agentes LangGraph​

O passo 7: Conceder acesso ao seu agente implantado​

O passo 8: Testar o agente de ponta a ponta​

Solução de problemas​

Limpar​

Os passos seguintes​