Azure AI Fondry Agenst ServiceのSDKについてまとめてみた

{ 
    "id": "msg_hogehoge", 
    "object": "thread.message", 
    "created_at": 1749187901, 
    "assistant_id": "asst_hogehoge", 
    "thread_id": "thread_hogehoge", 
    "run_id": "run_hogehoge", 
    "role": "assistant", 
    "content": [ 
        { 
            "type": "text", 
            "text": { "value": "このドキュメントにはhogehogeと書いてあります【4:1†source】。" }, 
            "annotations": [ 
                { 
                    "type": "file_citation", 
                    "text": "【4:1†source】", 
                    "start_index": 331, 
                    "end_index": 343, 
                    "file_citation": { "file_id": "assistant-hogehoge" } 
                }
            ]
        }
    ], 
    "attachments": [], 
    "metadata": {}
}

import asyncio
import os
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential
from azure.ai.agents.models import (
    TruncationObject, TruncationStrategy,
    ThreadMessageOptions,
    MessageRole, MessageInputTextBlock,
)

async def main():
    PROJECT_ENDPOINT = "Azure AI Fonudryのプロジェクのエンドポイント"
    AGENT_ID = "エージェントID"
    MODEL_ID = "モデルID"
    # エージェント全体の出力を決めるシステムインストラクションの設定
    INSTRUCTIONS = "日本語で会話を行うAIエージェントです。ユーザーの質問に対して、適切な回答を提供します。"
    # スレッドのRUN単位でのエージェントの出力を決める追加のインストラクション
    ADDITTIONAL_INSTRUCTIONS = "会話相手が子供であることを考慮して、フランクに応答してください。"

    # エージェントの追加メッセージとパラメータ
    temperature = 1.0
    top_p = 0.1
    max_prompt_tokens = 16384
    max_completion_tokens = 4096
    polling_interval = 2
    response_format = "auto" # フォーマットは現状autoのみ
    last_messages = 4

    # Projectクライアントの作成
    # noinspection PyTypeChecker
    project_client = AIProjectClient(
        endpoint=PROJECT_ENDPOINT,
        credential=DefaultAzureCredential()
    )

    # エージェント取得
    agent_client = await project_client.agents.get_agent(agent_id=AGENT_ID)
    agent_client.instructions = INSTRUCTIONS
    
    # スレッド作成
    thread = await project_client.agents.threads.create()
    
    # スレッドにメッセージを追加
    contents = [
        MessageInputTextBlock(
            {
                "type": "text",
                "text": "私の名前はジョンです"
            }
        )
    ]
    message = await project_client.agents.messages.create(
        thread_id=thread.id,
        role=MessageRole.USER,
        content=contents,
    )

    # トランケーション戦略の設定
    truncation_strategy = TruncationObject(
        type=TruncationStrategy.LAST_MESSAGES,
        last_messages=last_messages
    )

    # 会話を追加メッセージスレッドの最後列に追加
    additional_messages = [
        {
            "role": MessageRole.AGENT,
            "content": "こんにちは！"
        },
        {
            "role": MessageRole.USER,
            "content": "私の名前をおぼえていますか?"
        }
    ]
    additional_thread_messages = []
    for msg in additional_messages:
        thread_message = ThreadMessageOptions(
            role=msg.get("role", "user"),
            content=msg.get("content", "")
        )
        additional_thread_messages.append(thread_message)

    # RUN実行し、エージェントメッセージを送信
    run = await project_client.agents.runs.create_and_process(
        thread_id=thread.id,
        agent_id=agent_client.id,
        model=MODEL_ID,
        additional_instructions=ADDITTIONAL_INSTRUCTIONS,
        additional_messages=additional_thread_messages,
        temperature=temperature,
        top_p=top_p,
        max_prompt_tokens=max_prompt_tokens,
        max_completion_tokens=max_completion_tokens,
        polling_interval=polling_interval,
        response_format=response_format,
        truncation_strategy=truncation_strategy
    )
    print(f"Run result: [{run}]")
    print(
        f"thread id: [{run.thread_id}], "
        f"status: [{run.status}], "
        f"incomplete details: [{run.incomplete_details}], "
        f"cancelled: [{run.cancelled_at}], "
        f"failed: [{run.failed_at}], "
        f"last error: [{run.last_error}]"
    )
    print(f"Run result usage: [{run.usage}]")

    # Threadのメッセージを降順に取得して表示
    print("-----------------------------------------------------------------")
    messages = project_client.agents.messages.list(thread_id=thread.id)
    idx = 1
    async for message in messages:
        message_text_details = message.content[0].get("text", {})
        text_value = message_text_details.get("value")
        role = message.role.value
        print(f"[{idx}]. Role: [{role}], Text: [{text_value}], message keys[{message.keys()}]")
        idx += 1
    print(f"-----------------------------------------------------------------\n")

    # RUNをストリーミングで実行
    print("Starting run streaming...")
    stream = await project_client.agents.runs.stream(
        thread_id=thread.id,
        agent_id=agent_client.id,
        model=MODEL_ID,
        additional_instructions=ADDITTIONAL_INSTRUCTIONS,
        additional_messages=additional_thread_messages,
        temperature=temperature,
        top_p=top_p,
        max_prompt_tokens=max_prompt_tokens,
        max_completion_tokens=max_completion_tokens,
        response_format=response_format,
        truncation_strategy=truncation_strategy
    )

    # ストリーミングで送信する場合
    event_count = 0
    async for event in stream.event_handler:
        event_count += 1
        # イベントの種類に応じて処理を分岐
        # イベントの種類は以下の通り
        # [thread.run.created]: RUN作成完了, data:[<class 'azure.ai.agents.models._models.ThreadRun'>]        
        # [thread.run.queued]: QUEUEの送信完了 data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
        # [thread.run.in_progress]: RUNの実行中, data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
        # [thread.run.step.created]: STEPの作成, data:[<class 'azure.ai.agents.models._models.RunStep'>]
        # [thread.run.step.in_progress]: STEPの実行中, data:[<class 'azure.ai.agents.models._models.RunStep'>] 
        # [thread.message.created]: MESSAGEの作成完了, data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
        # [thread.message.in_progress]: MESSAGEの処理中, data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
        # [thread.message.delta]: ストリーミング中のメッセージの一部を送信, data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
        # [thread.message.completed]: ストリーミングのメッセージ送信完了通知, data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
        # [thread.run.step.completed]: STEPの完了, data:[<class 'azure.ai.agents.models._models.RunStep'>]
        # [thread.run.completed]: RUNの完了, data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
        # [done]: ストリーミングの完了通知, data:[<class 'str'>]
        event_status = event[0]
        event_data = event[1]
        print(f"[{event_count}], Event: type[{type(event)}] stats:[{event_status}], data:[{type(event_data)}]")
        # ストリーミング中のメッセージの一部を表示
        if event_status == "thread.message.delta":
            content = event_data.delta.content
            text_value = content[0].get("text", {}).get("value", "")
            print(f"    Delta message content: [{text_value}]")
        # ストリームのの完了後のメッセージの全体を表示
        elif event_status == "thread.message.completed":
            content = event_data.content
            text_value = content[0].get("text", {}).get("value", "")
            print(f"    Message Completed Content [{event_data.content}]")
        # Runの実行完了後のトークン使用量を表示
        elif event_status == "thread.run.completed":
            usage = event_data.usage
            print(f"    Run completed, Usage: [{usage}]")
        elif event_status == "done":
            print(f"    Stream completed.{event_data}")

    print("Run streaming completed.")
    await project_client.close()

if __name__ == "__main__":
    asyncio.run(main())

Run result: [{'id': 'run_hogehoge', 'object': 'thread.run', 'created_at': 1753512238, 'assistant_id': 'asst_hogehoge', 'thread_id': 'thread_hogehoge', 'status': 'completed', 'started_at': 1753512238, 'expires_at': None, 'cancelled_at': None, 'failed_at': None, 'completed_at': 1753512239, 'required_action': None, 'last_error': None, 'model': 'gpt-4o-mini', 'instructions': ' 会話相手が子供であることを考慮して、フランクに応答してください。', 'tools': [{'type': 'file_search'}], 'tool_resources': {}, 'metadata': {}, 'temperature': 1.0, 'top_p': 0.1, 'max_completion_tokens': 4096, 'max_prompt_tokens': 16384, 'truncation_strategy': {'type': 'last_messages', 'last_messages': 4}, 'incomplete_details': None, 'usage': {'prompt_tokens': 702, 'completion_tokens': 26, 'total_tokens': 728, 'prompt_token_details': {'cached_tokens': 0}}, 'response_format': 'auto', 'tool_choice': 'auto', 'parallel_tool_calls': True}]
thread id: [thread_hogehoge], status: [RunStatus.COMPLETED], incomplete details: [None], cancelled: [None], failed: [None], last error: [None]
Run result usage: [{'prompt_tokens': 702, 'completion_tokens': 26, 'total_tokens': 728, 'prompt_token_details': {'cached_tokens': 0}}]
-----------------------------------------------------------------
[1]. Role: [assistant], Text: [うん、ジョンだよね！覚えてるよ！他に何か話したいことある？], message keys[dict_keys(['id', 'object', 'created_at', 'assistant_id', 'thread_id', 'run_id', 'role', 'content', 'attachments', 'metadata'])]
[2]. Role: [user], Text: [私の名前をおぼえていますか?], message keys[dict_keys(['id', 'object', 'created_at', 'assistant_id', 'thread_id', 'run_id', 'role', 'content', 'attachments', 'metadata'])]
[3]. Role: [assistant], Text: [こんにちは！], message keys[dict_keys(['id', 'object', 'created_at', 'assistant_id', 'thread_id', 'run_id', 'role', 'content', 'attachments', 'metadata'])]
[4]. Role: [user], Text: [私の名前はジョンです], message keys[dict_keys(['id', 'object', 'created_at', 'assistant_id', 'thread_id', 'run_id', 'role', 'content', 'attachments', 'metadata'])]
-----------------------------------------------------------------

Starting run streaming...
[1], Event: type[<class 'tuple'>] stats:[thread.run.created], data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
[2], Event: type[<class 'tuple'>] stats:[thread.run.queued], data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
[3], Event: type[<class 'tuple'>] stats:[thread.run.in_progress], data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
[4], Event: type[<class 'tuple'>] stats:[thread.run.step.created], data:[<class 'azure.ai.agents.models._models.RunStep'>]
[5], Event: type[<class 'tuple'>] stats:[thread.run.step.in_progress], data:[<class 'azure.ai.agents.models._models.RunStep'>] 
[6], Event: type[<class 'tuple'>] stats:[thread.message.created], data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
[7], Event: type[<class 'tuple'>] stats:[thread.message.in_progress], data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
[8], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [う]
[9], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [ん]
[10], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [、]
[11], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [君]
[12], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [の]
[13], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [名前]
[14], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [は]
[15], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [ジョ]
[16], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [ン]
[17], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [だ]
[18], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [よ]
[19], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [ね]
[20], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [！]
[21], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [他]
[22], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [に]
[23], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [何]
[24], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [か]
[25], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [話]
[26], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [した]
[27], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [い]
[28], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [こと]
[29], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [が]
[30], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [ある]
[31], Event: type[<class 'tuple'>] stats:[thread.message.delta], data:[<class 'azure.ai.agents.models._patch.MessageDeltaChunk'>]
    Delta message content: [？]
[32], Event: type[<class 'tuple'>] stats:[thread.message.completed], data:[<class 'azure.ai.agents.models._patch.ThreadMessage'>]
    Message Completed Content [[{'type': 'text', 'text': {'value': 'うん、君の名前はジョンだよね！他に何か話したいことがある？', 'annotations': []}}]]
[33], Event: type[<class 'tuple'>] stats:[thread.run.step.completed], data:[<class 'azure.ai.agents.models._models.RunStep'>]
[34], Event: type[<class 'tuple'>] stats:[thread.run.completed], data:[<class 'azure.ai.agents.models._models.ThreadRun'>]
    Run completed, Usage: [{'prompt_tokens': 719, 'completion_tokens': 26, 'total_tokens': 745, 'prompt_token_details': {'cached_tokens': 0}}]
[35], Event: type[<class 'tuple'>] stats:[done], data:[<class 'str'>]
    Stream completed.[DONE]
Run streaming completed.

# AIProjectClientの作成
endpoint = "https://<ai-services-account-name>.services.ai.azure.com/api/projects/<project_name>"
credential = DefaultAzureCredential()
project_client = AIProjectClient(endpoint, credential)

# AIProjectClientのagentsフィールドはAgentsClient型で、AgentやVectorStoreへの操作が可能

# Agentを取得
agent = await self.project_client.agents.get_agent(agent_id="hogehoge")
# Agentを更新
agent = await self.project_client.agents.update_agent(
    agent_id="hogehoge",
    tools=tools.definitions,
    tool_resources=tool_resources
)

# スレッドを作成
thread = await self.project_client.agents.threads.create()
# スレッドを取得
thread = await self.project_client.agents.threads.get(thread_id="hogehoge")

from azure.ai.agents.aio import AgentsClient
from azure.identity.aio import DefaultAzureCredential

client = AgentsClient(
    endpoint="https://your-resource-endpoint",
    credential=DefaultAzureCredential()
)

message = MessageInputTextBlock(
    {
        "type": "text",
        "text": "Hello"
    }
)

message = MessageInputImageUrlBlock(
     {
       "type": "image_url",
       "image_url": {
          "url": "https://hogehoge.com/hoge.jpg"
      }
    }
)

# str型でロールとコンテンツを設定する場合
thread_message_options = ThreadMessageOptions(
    role="user",
    content="hello"
)

# MessageRoleとlist[MessageInputContentBlock]でロールとコンテンツを設定する場合
thread_message_options = ThreadMessageOptions(
    role=MessageRole.USER,
    content=[
        message = MessageInputTextBlock(
            {
                "type": "text",
                "text": "Hello"
            }
        )
    ]
)

# AUTO: 設定されているコンテキストウィンドウを超過した場合に自動削除
truncation_strategy = TruncationObject(
    type=TruncationStrategy.AUTO
)
# LAST_MESSAGES: 最後のlast_messages数のメッセージを保持する
truncation_strategy = TruncationObject(
    type=TruncationStrategy.LAST_MESSAGES,
    last_messages=2
)

# MessageTextContentの場合の中身
# 【6:11†source】は生成したコンテンツがFile Search Toolで検索にヒットしたファイルのうち、どのファイルをもとに生成したかを示す
{
  'type': 'text',
  'text': {
    'value': 'Hello【6:11†source】。', 
    'annotations': [
      {
        'type': 'file_citation',
        'text': '【6:11†source】',
        'start_index': 85,
        'end_index': 98,
        'file_citation': {
          'file_id': 'assistant-hogehoge'
        }
      }
    ]
  }
}

# MessageTextFileCitationAnnotationの場合の中身
{
  'type': 'file_citation',
  'text': '【6:11†source】',
  'start_index': 85,
  'end_index': 98,
  'file_citation': {'file_id': 'assistant-hogehoge'}
}

class Tool(ABC, Generic[ToolDefinitionT]):
    """ 
    An abstract class representing a tool that can be used by an agent.
    """

    @property
    @abstractmethod
    def definitions(self) -> List[ToolDefinitionT]:
        """Get the tool definitions."""

    @property
    @abstractmethod
    def resources(self) -> ToolResources:
        """Get the tool resources."""

    @abstractmethod
    def execute(self, tool_call: Any) -> Any:
        """
        Execute the tool with the provided tool call.

        :param Any tool_call: The tool call to execute.
        :return: The output of the tool operations.
        """

# ファイル検索ツールのインスタンス生成
file_search_tool = FileSearchTool(vector_store_ids=["vector_store_hogehoge"])

# Agentにファイル検索ツールからツール定義情報(definitions)と接続リソース情報(resources)を渡します
project_client = AIProjectClient(endpoint, credential)
agent = project_client.agents.update_agent(
    agent_id="agent_hogehoge",
    tools=file_search_tool.definitions,
    tool_resources=file_search_tool.resources
)

元文書: [1000トークンの長い文書] 
  - チャンク1: [800トークン] 
  - チャンク2: [200トークン] 

文書: "太郎は学校に行った。そこで友達の花子に会った。花子は図書館で本を読んでいた。" 
max_chunk_size_tokens=20, chunk_overlap_tokens=10の場合: 
  - チャンク1: "太郎は学校に行った。そこで友達の花子に会った。" (20トークン) 
  - チャンク2: "そこで友達の花子に会った。花子は図書館で本を読んでいた。" (10トークン重複 + 10トークン新規) 

# type: autoを指定
chunking_strategy = VectorStoreChunkingStrategyRequest(type="auto") 

# JSON形式のマッピングから指定も可能
mapping = {"type": "auto"} 
chunking_strategy = VectorStoreChunkingStrategyRequest(mapping)

# type: staticを指定するときはVectorStoreStaticChunkingStrategyRequestを使用
VectorStoreStaticChunkingStrategyOptions(
    max_chunk_size_tokens=max_chunk_size_tokens,
    chunk_overlap_tokens=chunk_overlap_tokens
)
chunking_strategy = VectorStoreStaticChunkingStrategyRequest(
    static=VectorStoreStaticChunkingStrategyOptions
)