Microsoft Agent Framework開発入門 #Azure – Qiita

import asyncio
from dataclasses import dataclass
from typing_extensions import Never

from agent_framework import (
    WorkflowBuilder,
    WorkflowContext,
    Executor,
    RequestInfoExecutor,
    RequestInfoEvent,
    RequestInfoMessage,
    RequestResponse,
    WorkflowOutputEvent,
    WorkflowStatusEvent,
    WorkflowRunState,
    handler,
)

# ---------- 型定義 ----------
@dataclass
class ApprovalRequest(RequestInfoMessage):
    """承認リクエスト（RequestInfoMessage を継承）"""
    prompt: str = ""
    attempt_count: int = 1

# ---------- 承認管理 Executor ----------
class ApprovalManager(Executor):
    """承認プロセスを管理するExecutor"""

    def __init__(self, id: str | None = None):
        super().__init__(id=id or "approval_manager")

    @handler
    async def start(self, _: str, ctx: WorkflowContext[ApprovalRequest]) -> None:
        """最初の承認リクエストを送信"""
        print("[ApprovalManager] 承認プロセスを開始します")
        await ctx.send_message(ApprovalRequest(
            prompt="レポート公開の承認をお願いします (y/n)",
            attempt_count=1
        ))

    @handler
    async def handle_response(
        self,
        feedback: RequestResponse[ApprovalRequest, str],
        ctx: WorkflowContext[ApprovalRequest, str],
    ) -> None:
        """承認レスポンスを処理"""
        reply = (feedback.data or "").strip().lower()
        request_data = feedback.original_request
        
        print(f"[ApprovalManager] 試行 {request_data.attempt_count}: 応答='{reply}'")
        
        if reply in ("y", "yes", "1", "ok", "承認"):
            # 承認された場合
            print(f"[ApprovalManager] 承認されました！(試行回数: {request_data.attempt_count})")
            await ctx.yield_output(f"レポートが正常に公開されました (試行回数: {request_data.attempt_count})")
            return
        
        # 却下された場合、再試行
        next_attempt = request_data.attempt_count + 1
        print(f"[ApprovalManager] 却下されました。{next_attempt} 回目の承認をリクエストします...")
        
        await ctx.send_message(ApprovalRequest(
            prompt=f"レポート公開の承認をお願いします (y/n) - {next_attempt} 回目の試行",
            attempt_count=next_attempt
        ))

# ---------- ワークフロー構築 ----------
def build_workflow():
    """承認ワークフローを構築"""
    approval_manager = ApprovalManager(id="approval_manager")
    request_info = RequestInfoExecutor(id="request_info")
    
    workflow = (
        WorkflowBuilder()
        .set_start_executor(approval_manager)
        .add_edge(approval_manager, request_info)  # 承認リクエスト送信
        .add_edge(request_info, approval_manager)  # 承認レスポンス受信
        .build()
    )
    
    return workflow

# ---------- メイン実行 ----------
async def main():
    """メイン実行関数（公式サンプルのパターンに基づく）"""
    workflow = build_workflow()
    
    pending_responses: dict[str, str] | None = None
    completed = False
    workflow_output: str | None = None
    
    while not completed:
        # 初回は run_stream、2回目以降は send_responses_streaming
        stream = (
            workflow.send_responses_streaming(pending_responses) 
            if pending_responses 
            else workflow.run_stream("start")
        )
        
        # イベントを収集
        events = [event async for event in stream]
        pending_responses = None
        
        # リクエストと出力を収集
        requests: list[tuple[str, str, int]] = []  # (request_id, prompt, attempt_count)
        
        for event in events:
            if isinstance(event, RequestInfoEvent) and isinstance(event.data, ApprovalRequest):
                # 承認リクエストイベント
                requests.append((
                    event.request_id, 
                    event.data.prompt, 
                    event.data.attempt_count
                ))
            elif isinstance(event, WorkflowOutputEvent):
                # ワークフロー出力（完了）
                workflow_output = str(event.data)
                completed = True
        
        # 状態変化を表示（デバッグ用）
        pending_status = any(
            isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS
            for e in events
        )
        idle_with_requests = any(
            isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
            for e in events
        )
        
        if pending_status:
            print("[状態] IN_PROGRESS_PENDING_REQUESTS (リクエスト処理中)")
        if idle_with_requests:
            print("[状態] IDLE_WITH_PENDING_REQUESTS (人間の入力待ち)")
        
        # 承認リクエストがある場合、ユーザーに入力を求める
        if requests and not completed:
            responses: dict[str, str] = {}
            
            for req_id, prompt, attempt_count in requests:
                print(f"\n{'='*60}")
                print(f"承認リクエスト (試行 {attempt_count})")
                print(f"{'='*60}")
                print(f"{prompt}")
                
                answer = input("承認しますか？ (y/n/exit): ").strip().lower()
                
                if answer == "exit":
                    print("ユーザーによって処理が中断されました")
                    return
                
                responses[req_id] = answer
                print(f"入力結果: {'承認' if answer in ('y', 'yes', '1', 'ok', '承認') else '却下'}")
            
            pending_responses = responses
    
    # 最終結果を表示
    print(f"\n{'='*70}")
    print(f"最終結果: {workflow_output}")
    print(f"{'='*70}")
    
    # ワークフロー図を表示
    print(f"\n{'='*50}")
    print("ワークフロー図 (Mermaid)")
    print(f"{'='*50}")
    from agent_framework import WorkflowViz
    viz = WorkflowViz(workflow)
    print(viz.to_mermaid())

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


# ---------- 出力 ----------
# [ApprovalManager] 承認プロセスを開始します
# [状態] IN_PROGRESS_PENDING_REQUESTS (リクエスト処理中)
# [状態] IDLE_WITH_PENDING_REQUESTS (人間の入力待ち)

# ============================================================
# 承認リクエスト (試行 1)
# ============================================================
# レポート公開の承認をお願いします (y/n)
# 承認しますか？ (y/n/exit): n
# 入力結果: 却下
# [ApprovalManager] 試行 1: 応答='n'
# [ApprovalManager] 却下されました。2 回目の承認をリクエストします...
# [状態] IN_PROGRESS_PENDING_REQUESTS (リクエスト処理中)
# [状態] IDLE_WITH_PENDING_REQUESTS (人間の入力待ち)

# ============================================================
# 承認リクエスト (試行 2)
# ============================================================
# レポート公開の承認をお願いします (y/n) - 2 回目の試行
# 承認しますか？ (y/n/exit): y
# 入力結果: 承認
# [ApprovalManager] 試行 2: 応答='y'
# [ApprovalManager] 承認されました！(試行回数: 2)

# ======================================================================
# 最終結果: レポートが正常に公開されました (試行回数: 2)
# ======================================================================