【Azure AI Foundry：第2回】知能のステート管理：Threadによる会話履歴の自動永続化

【新連載：Azure AI Foundryとマルチエージェント】

前回（第1回）：【Azure AI Foundry：第1回】クラウド知能への接続：SDKのセットアップとAIProjectClient

※本連載は、Azure AI Foundry SDK (azure-ai-projects) を用い、実務で安全・強力に動作する「エンタープライズ仕様の自律チームエージェント」を構築するステップバイステップのハンズオン連載です。

1. イントロダクション（今回の到達目標）

前回は、Azure AI Foundry へのセキュアなゲートウェイである AIProjectClient を初期化し、Entra ID 認証を用いてクラウド上のモデルと対話する基盤を構築しました。

しかし、前回の実装はユーザーからの入力に対して1回限りの応答を返す「ステートレス（一問一答）」な状態でした。エージェントがユーザーの文脈を理解し、複雑なタスクを段階的にクリアするためには、過去の会話履歴や推論プロセスを記憶し続ける「ステートフル（状態維持）」な設計が不可欠です。

今回は、会話の履歴をクラウド側で自動的に永続化・管理するステート機能、「Thread（スレッド）」と「Agent（エージェント）」の概念を学び、同じスレッドに複数のメッセージを連続して送信することで、過去の会話文脈が正しく引き継がれて対話が継続する仕組みを構築することを目標とします。

前提条件（Prerequisites）

動作確認済みライブラリ: azure-ai-projects>=2.0.0
事前準備：前回の環境変数 AZURE_AI_PROJECT_ENDPOINT の設定、および az login による CLI 認証が完了していること。

2. クラウド側で記憶を保持する「Thread」の設計

従来の簡易的なチャットアプリやAIシステムでは、会話の履歴（コンテキスト）を維持するために、以下のような力業が用いられていました。

過去のやり取りをすべてクライアント側のメモリやローカルデータベースに保存しておく。
新しい質問を投げるたびに、過去の全会話履歴をプロンプトの前方に合体（累積）させてLLMへ毎回再送信する。

この方法は、クライアント側のコードが複雑になり、かつ会話が長くなるほどネットワークの通信量やトークン消費量が雪だるま式に膨らんでいくという弱点があります。

Azure AI Foundry の Agent サービスが提供する「Thread」は、この課題をクラウド側でエレガントに解決します。

会話履歴はすべてAzureのクラウド上に暗号化されて永続化されるため、クライアント側は単に「どの会話セッションか」を指し示す thread_id だけを保持すればよくなります。新しい発言（Message）をスレッドに追加して実行（Run）をトリガーするだけで、LLMは過去のすべての文脈を自動的にロードして推論を行います。

3. 【実践】ThreadとAgentを用いた対話の継続

実際にエージェントとスレッドをクラウド上に作成し、自己紹介を行った直後に「自分の名前は何か」を問いかけ、正しく文脈が永続化されているかを検証する完全な Python スクリプトを実行してみましょう。

# 動作確認済みライブラリバージョン: azure-ai-projects==2.0.0
import os
import time
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient

def main():
    # 1. クライアントの初期化
    endpoint = os.environ.get("AZURE_AI_PROJECT_ENDPOINT")
    if not endpoint:
        print("Error: 環境変数 AZURE_AI_PROJECT_ENDPOINT が設定されていません。")
        return

    credential = DefaultAzureCredential()
    
    # リソースのクリーンアップを確実にするため context manager を使用します
    with AIProjectClient(endpoint=endpoint, credential=credential) as project_client:
        
        # 2. Azure上にエージェントをデプロイ
        print("🤖 エージェントを作成中...")
        agent = project_client.agents.create_agent(
            model="gpt-4o",  # ポータル上でデプロイした「モデルのデプロイ名」を指定
            name="anatomy-helper-agent",
            instructions="あなたは親切なAIアシスタントです。ユーザーの名前を覚えたら、それ以降は常に名前で呼びかけてください。"
        )
        print(f"✅ エージェント作成完了 (ID: {agent.id})")

        # 3. 会話セッション (Thread) の作成
        print("\n💬 新しい会話スレッドを作成中...")
        thread = project_client.agents.create_thread()
        print(f"✅ スレッド作成完了 (Thread ID: {thread.id})")

        # 4. 最初の発言をスレッドに追加（自己紹介）
        print("\n➡️ ユーザー: 私の名前は太郎です。")
        project_client.agents.create_message(
            thread_id=thread.id,
            role="user",
            content="私の名前は太郎です。新しくエージェントの勉強を始めました。"
        )

        # 5. 実行のトリガー（同期メソッドを用いて処理完了を待機）
        print("⚡ システム: エージェントを実行中...")
        run = project_client.agents.create_and_process_run(
            thread_id=thread.id,
            assistant_id=agent.id
        )
        print(f"✅ 実行ステータス: {run.status}")

        # 6. 2番目の発言を追加（記憶の確認）
        print("\n➡️ ユーザー: 私の名前がわかりますか？")
        project_client.agents.create_message(
            thread_id=thread.id,
            role="user",
            content="私の名前がわかりますか？"
        )

        print("⚡ システム: エージェントを再実行中...")
        run = project_client.agents.create_and_process_run(
            thread_id=thread.id,
            assistant_id=agent.id
        )
        print(f"✅ 実行ステータス: {run.status}")

        # 7. スレッド内のメッセージ全履歴を取得して表示
        print("\n📥 会話履歴の取得中...")
        messages = project_client.agents.list_messages(thread_id=thread.id)
        
        print("\n=== スレッド全体の会話ログ ===")
        # メッセージは降順（新しい順）で返されるため、逆順にして表示します
        for msg in reversed(messages.data):
            # contentオブジェクトからテキストデータを抽出
            text_content = "".join([part.text.value for part in msg.content if hasattr(part, 'text')])
            print(f"[{msg.role.upper()}]: {text_content}")
        print("==============================")

        # テスト用エージェントの削除（任意）
        project_client.agents.delete_agent(agent.id)
        print(f"\n🧹 リソースをクリーンアップしました。")

if __name__ == "__main__":
    main()

コードの挙動解説

create_agent: Azure上に推論ルールを持ったエージェントの実体を作成します。
create_thread: 会話データを保存するユニークな箱を用意します。
create_and_process_run: 単に推論を依頼するだけでなく、エージェントが Thread の過去データを自律的にスキャンし、新しい応答を生成して自動で Thread に追加するまでの一連の「推論ループ」を同期的に待機します。

このプログラムを実行すると、2回目の「私の名前がわかりますか？」という質問に対し、以前送った「太郎です」という情報をAzure側の Thread からAIが自動的に読み取り、「はい、太郎さんですね！」と正確に答える様子が確認できます。

4. まとめと次回の展望

今回は、会話の状態（ステート）をクラウドの Thread に永続化させることで、エージェントに「持続する記憶」を実装することに成功しました。

しかし、いくら会話の流れを覚えていても、エージェントはまだ「自分のトレーニングデータに含まれていない社内のプライベートなデータやファイル」の内容を知ることはできません。本物の実務用エージェントにするためには、社外秘の情報が詰まったドキュメントを検索し、それに基づいて回答する能力が必要です。

次回、第3回。
Azure上にアップロードした独自のナレッジベースから、セキュリティ境界を保ったままセマンティック（意味的）検索を行い、回答の根拠とさせる、「Azure AI SearchとのRAGエージェント構築」に進みます。

知能が、社内の膨大なデータという「外部知識」とシームレスに融合する瞬間へと進みましょう。