Azure Cosmos DB, un servizio di database distribuito globale di Microsoft

Installare i pacchetti

dotnet add package Microsoft.Agents.AI.CosmosNoSql --prerelease
dotnet add package Azure.Identity

Configurare la cronologia delle chat di Cosmos DB

Usare l'estensione managed-identity per collegare CosmosChatHistoryProvider a ChatClientAgentOptions.

using Azure.Identity;
using Microsoft.Agents.AI;

var options = new ChatClientAgentOptions
{
    ChatOptions = new() { Instructions = "You are a helpful assistant." }
}.WithCosmosDBChatHistoryProviderUsingManagedIdentity(
    accountEndpoint: Environment.GetEnvironmentVariable("AZURE_COSMOS_ENDPOINT")!,
    databaseId: Environment.GetEnvironmentVariable("AZURE_COSMOS_DATABASE_NAME")!,
    containerId: Environment.GetEnvironmentVariable("AZURE_COSMOS_CONTAINER_NAME")!,
    tokenCredential: new DefaultAzureCredential());

AIAgent agent = chatClient.AsAIAgent(options);

L'inizializzatore di stato predefinito crea un ID conversazione. Specificare un CosmosChatHistoryProvider.State inizializzatore quando l'applicazione necessita di conversazioni, tenant e routing utente espliciti. Quando sono presenti ID tenant e utente, il provider usa una chiave di partizione gerarchica.

Installare il pacchetto

pip install agent-framework-azure-cosmos --pre

ConfigurareCosmosHistoryProvider

Il provider Python accetta una credenziale Azure o una chiave dell'account e usa come session_id chiave di partizione.

# 1. Create an Azure credential and a CosmosHistoryProvider for agent context
async with (
    AzureCliCredential() as credential,
    CosmosHistoryProvider(
        endpoint=cosmos_endpoint,
        database_name=cosmos_database_name,
        container_name=cosmos_container_name,
        credential=cosmos_key or credential,
    ) as history_provider,
    # 2. Create an agent that uses Cosmos for persisted conversation history.
    Agent(
        client=FoundryChatClient(
            project_endpoint=project_endpoint,
            model=model,
            credential=credential,
        ),
        name="CosmosHistoryAgent",
        instructions="You are a helpful assistant that remembers prior turns.",
        context_providers=[history_provider],
        default_options={"store": False},
    ) as agent,
):
    # 3. Create a session (session_id is used as the partition key).
    session = agent.create_session()

    # 4. Run a multi-turn conversation; history is persisted by CosmosHistoryProvider.
    response1 = await agent.run("My name is Ada and I enjoy distributed systems.", session=session)
    print(f"Assistant: {response1.text}")

    response2 = await agent.run("What do you remember about me?", session=session)
    print(f"Assistant: {response2.text}")
    print(f"Container: {history_provider.container_name}")

Rendere persistente il serializzato AgentSession nell'archiviazione di applicazioni attendibili quando i client devono ripristinare lo stesso identificatore di sessione in un secondo momento.

Prerequisites

• Un account e un database di Azure Cosmos DB.
• Un progetto Microsoft Foundry con distribuzioni di modelli di chat e di embedding.
• Accesso alle identità di Azure a entrambe le risorse.

Installare i pacchetti

pip install agent-framework-azure-cosmos-memory agent-framework-foundry --pre

Configurare il provider di memoria

Lo stesso progetto Foundry può fornire il modello di chat, incorporamenti e modello di estrazione della memoria. Collega il provider attraverso context_providers.

def _build_agent(provider: CosmosMemoryContextProvider, credential: DefaultAzureCredential) -> Agent:
    """Build an agent that uses the memory provider and the same Foundry endpoint for chat."""
    return Agent(
        client=FoundryChatClient(
            project_endpoint=os.environ["FOUNDRY_ENDPOINT"],
            model=os.getenv("CHAT_MODEL", "gpt-4o-mini"),
            credential=credential,
        ),
        name="Memory Assistant",
        instructions="You are a helpful assistant with long-term memory about the user.",
        context_providers=[provider],
    )


async def user_scoped_memory() -> None:
    """Memory scoped to a stable user id, so it persists across sessions and threads."""
    credential = DefaultAzureCredential()
    provider = CosmosMemoryContextProvider(
        cosmos_endpoint=os.environ["COSMOS_ENDPOINT"],
        foundry_endpoint=os.environ["FOUNDRY_ENDPOINT"],
        embedding_model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-large"),
        chat_model=os.getenv("CHAT_MODEL", "gpt-4o-mini"),
        credential=credential,
    )
    agent = _build_agent(provider, credential)

    async with provider:
        session = agent.create_session()
        # Provider state is scoped by source id; set a stable user id there so memory
        # persists across sessions rather than being limited to this one.
        session.state.setdefault(provider.source_id, {})["user_id"] = "alice"
        first = await agent.run("I love hiking and I'm allergic to peanuts.", session=session)
        print("Assistant:", first.text)

        # A brand-new session for the same user still recalls the earlier facts.
        new_session = agent.create_session()
        new_session.state.setdefault(provider.source_id, {})["user_id"] = "alice"
        recall = await agent.run("What do you remember about me?", session=new_session)
        print("Assistant:", recall.text)

        # Let background extraction finish and persist before the client closes.
        await provider.flush()

Un user_id stabile mantiene la memoria disponibile tra sessioni e thread. In assenza di questo, il provider limita la memoria all'ID di sessione corrente.

Elaborazione della memoria

L'estrazione della memoria viene eseguita in background dopo ogni turno. Usare il provider come gestore di contesto asincrono o chiamare flush() prima della chiusura, in modo che le estrazioni in sospeso vengano completate prima che i client vengano chiusi.

Il provider supporta anche richieste di estrazione personalizzate, frequenza del processore, soglie di confidenza, tipi di memoria e limiti di recupero.