Azure Cosmos DB

パッケージをインストールする

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

Cosmos DB チャット履歴の構成

マネージド ID 拡張機能を使用して、CosmosChatHistoryProviderに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);

既定の状態初期化子は、会話 ID を作成します。 アプリケーションで明示的な会話、テナント、ユーザー ルーティングが必要な場合は、 CosmosChatHistoryProvider.State 初期化子を指定します。 テナント ID とユーザー ID が存在する場合、プロバイダーは階層パーティション キーを使用します。

パッケージをインストールする

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

CosmosHistoryProvider のコンフィギュレーション

Python プロバイダーは、Azure資格情報またはアカウント キーのいずれかを受け入れ、パーティション キーとしてsession_idを使用します。

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

後でクライアントが同じセッション識別子を回復する必要がある場合は、シリアル化された AgentSession を信頼されたアプリケーション ストレージに保持します。

前提条件

• Azure Cosmos DB アカウントとデータベース。
• チャットと埋め込みモデルのデプロイを含むMicrosoft Foundry プロジェクト。
• Azure ID による両方のリソースへのアクセス

パッケージをインストールする

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

メモリ プロバイダーを構成する

同じ Foundry プロジェクトで、チャット モデル、埋め込み、メモリ抽出モデルを提供できます。 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()

安定した user_id により、セッションとスレッド間でメモリを使用できます。 1 つを指定しない場合、プロバイダーはメモリのスコープを現在のセッション ID に設定します。

メモリ処理

メモリ抽出は、ターンごとにバックグラウンドで実行されます。 非同期コンテキスト マネージャーとしてプロバイダーを使用するか、シャットダウン前に flush() を呼び出して、クライアントが閉じる前に保留中の抽出が完了します。

プロバイダーでは、カスタム抽出プロンプト、プロセッサの周期、信頼度のしきい値、メモリの種類、取得の制限もサポートされています。