# DAGベースのワークフローエンジン
from dataclasses import dataclass, field
from typing import Callable, Any
from enum import Enum
import time
import logging
 
logger = logging.getLogger(__name__)
 
class NodeType(Enum):
    LLM = "llm"           # LLM処理ノード
    TOOL = "tool"          # ツール実行ノード
    CONDITION = "condition" # 条件分岐ノード
    PARALLEL = "parallel"   # 並列実行ノード
    SUBWORKFLOW = "subworkflow"  # サブワークフローノード
 
class NodeStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"
 
@dataclass
class NodeExecution:
    """ノード実行の記録"""
    node_name: str
    status: NodeStatus
    start_time: float
    end_time: float = 0.0
    result: Any = None
    error: str = None
 
    @property
    def duration(self) -> float:
        return self.end_time - self.start_time
 
@dataclass
class WorkflowNode:
    name: str
    type: NodeType
    handler: Callable
    next_nodes: list[str] = field(default_factory=list)
    condition: Callable = None  # 条件分岐用
    retry_count: int = 0       # リトライ回数
    timeout: float = 30.0      # タイムアウト秒数
    description: str = ""      # ノードの説明
 
class WorkflowEngine:
    def __init__(self, name: str = "default"):
        self.name = name
        self.nodes: dict[str, WorkflowNode] = {}
        self.state: dict[str, Any] = {}
        self.execution_log: list[NodeExecution] = []
        self.hooks: dict[str, list[Callable]] = {
            "before_node": [],
            "after_node": [],
            "on_error": [],
        }
 
    def add_node(self, node: WorkflowNode):
        self.nodes[node.name] = node
 
    def add_hook(self, event: str, callback: Callable):
        """フックを追加"""
        if event in self.hooks:
            self.hooks[event].append(callback)
 
    def _execute_hooks(self, event: str, **kwargs):
        for hook in self.hooks.get(event, []):
            hook(**kwargs)
 
    def _execute_node(self, node: WorkflowNode) -> Any:
        """ノードを実行（リトライ付き）"""
        last_error = None
        for attempt in range(node.retry_count + 1):
            try:
                self._execute_hooks("before_node", node=node, state=self.state)
                result = node.handler(self.state)
                self._execute_hooks("after_node", node=node, state=self.state, result=result)
                return result
            except Exception as e:
                last_error = e
                logger.warning(f"ノード {node.name} 失敗 (試行 {attempt + 1}): {e}")
                if attempt < node.retry_count:
                    time.sleep(2 ** attempt)  # 指数バックオフ
 
        self._execute_hooks("on_error", node=node, error=last_error)
        raise last_error
 
    def run(self, start_node: str, initial_state: dict) -> dict:
        self.state = initial_state
        self.execution_log = []
        current = start_node
 
        while current:
            node = self.nodes[current]
            execution = NodeExecution(
                node_name=node.name,
                status=NodeStatus.RUNNING,
                start_time=time.time()
            )
            logger.info(f"実行中: {node.name} ({node.type.value})")
 
            try:
                if node.type == NodeType.CONDITION:
                    branch = node.condition(self.state)
                    current = branch
                    execution.status = NodeStatus.COMPLETED
                    execution.result = f"分岐先: {branch}"
                elif node.type == NodeType.PARALLEL:
                    results = self._run_parallel(node.next_nodes)
                    self.state["parallel_results"] = results
                    current = node.next_nodes[-1] if node.next_nodes else None
                    execution.status = NodeStatus.COMPLETED
                else:
                    result = self._execute_node(node)
                    self.state[f"{node.name}_result"] = result
                    current = node.next_nodes[0] if node.next_nodes else None
                    execution.status = NodeStatus.COMPLETED
                    execution.result = result
            except Exception as e:
                execution.status = NodeStatus.FAILED
                execution.error = str(e)
                logger.error(f"ノード {node.name} で致命的エラー: {e}")
                raise
            finally:
                execution.end_time = time.time()
                self.execution_log.append(execution)
 
        return self.state
 
    def get_execution_summary(self) -> str:
        """実行サマリーを取得"""
        lines = [f"ワークフロー '{self.name}' 実行結果:"]
        total_duration = 0
        for ex in self.execution_log:
            status_icon = "✓" if ex.status == NodeStatus.COMPLETED else "✗"
            lines.append(
                f"  {status_icon} {ex.node_name}: {ex.duration:.2f}s "
                f"({ex.status.value})"
            )
            total_duration += ex.duration
        lines.append(f"  合計時間: {total_duration:.2f}s")
        return "\n".join(lines)

# 条件分岐を含むワークフロー
import anthropic
 
client = anthropic.Anthropic()
 
def classify_inquiry(message: str) -> str:
    """問い合わせをLLMで分類"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=50,
        messages=[{
            "role": "user",
            "content": f"""以下の問い合わせを分類してください。
カテゴリ: billing, technical, general
1単語で回答: {message}"""
        }]
    )
    return response.content[0].text.strip().lower()
 
def handle_billing_inquiry(state: dict) -> str:
    """請求関連の対応"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        system="あなたは請求担当のサポートエージェントです。丁寧かつ正確に回答してください。",
        messages=[{
            "role": "user",
            "content": state["user_message"]
        }]
    )
    return response.content[0].text
 
def handle_technical_inquiry(state: dict) -> str:
    """技術関連の対応"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        system="あなたは技術サポートエージェントです。技術的な問題を解決してください。",
        messages=[{
            "role": "user",
            "content": state["user_message"]
        }]
    )
    return response.content[0].text
 
def handle_general_inquiry(state: dict) -> str:
    """一般問い合わせの対応"""
    response = client.messages.create(
        model="claude-haiku-3-20240307",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": state["user_message"]
        }]
    )
    return response.content[0].text
 
def generate_response(state: dict) -> str:
    """最終回答を整形"""
    # 分類結果に応じた結果を取得
    classification = state.get("classify_result", "general")
    handler_key = f"handle_{classification}_result"
    raw_response = state.get(handler_key, "回答を生成できませんでした")
 
    return f"""
【カテゴリ】{classification}
【回答】
{raw_response}
 
何かご不明な点がございましたらお気軽にお問い合わせください。
"""
 
def build_support_workflow():
    engine = WorkflowEngine(name="カスタマーサポート")
 
    # ノード1: 問い合わせ分類
    engine.add_node(WorkflowNode(
        name="classify",
        type=NodeType.LLM,
        handler=lambda state: classify_inquiry(state["user_message"]),
        next_nodes=["route"],
        description="問い合わせ内容をLLMで分類"
    ))
 
    # ノード2: ルーティング（条件分岐）
    engine.add_node(WorkflowNode(
        name="route",
        type=NodeType.CONDITION,
        handler=None,
        condition=lambda state: {
            "billing": "handle_billing",
            "technical": "handle_technical",
            "general": "handle_general"
        }.get(state["classify_result"], "handle_general"),
        description="分類結果に基づいてルーティング"
    ))
 
    # ノード3a: 請求対応
    engine.add_node(WorkflowNode(
        name="handle_billing",
        type=NodeType.LLM,
        handler=lambda state: handle_billing_inquiry(state),
        next_nodes=["respond"],
        retry_count=2,
        description="請求関連の問い合わせに回答"
    ))
 
    # ノード3b: 技術対応
    engine.add_node(WorkflowNode(
        name="handle_technical",
        type=NodeType.LLM,
        handler=lambda state: handle_technical_inquiry(state),
        next_nodes=["respond"],
        retry_count=2,
        description="技術的な問い合わせに回答"
    ))
 
    # ノード3c: 一般対応
    engine.add_node(WorkflowNode(
        name="handle_general",
        type=NodeType.LLM,
        handler=lambda state: handle_general_inquiry(state),
        next_nodes=["respond"],
        retry_count=1,
        description="一般的な問い合わせに回答"
    ))
 
    # ノード4: 回答生成
    engine.add_node(WorkflowNode(
        name="respond",
        type=NodeType.LLM,
        handler=lambda state: generate_response(state),
        next_nodes=[],
        description="最終回答を整形して返す"
    ))
 
    return engine
 
# 実行
workflow = build_support_workflow()
result = workflow.run("classify", {
    "user_message": "先月の請求額が間違っています"
})
print(workflow.get_execution_summary())

# LLMがワークフロー自体を生成するパターン
import json
 
class DynamicWorkflowBuilder:
    """LLMの判断でワークフローを動的に構築"""
 
    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.available_handlers = {}
 
    def register_handler(self, name: str, handler: Callable, description: str):
        """利用可能なハンドラーを登録"""
        self.available_handlers[name] = {
            "handler": handler,
            "description": description
        }
 
    def build_workflow(self, task_description: str) -> WorkflowEngine:
        """タスク記述からワークフローを自動生成"""
        handler_descriptions = "\n".join(
            f"- {name}: {info['description']}"
            for name, info in self.available_handlers.items()
        )
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{
                "role": "user",
                "content": f"""以下のタスクを処理するワークフローをJSON形式で設計してください。
 
タスク: {task_description}
 
利用可能なハンドラー:
{handler_descriptions}
 
出力形式:
{{
    "nodes": [
        {{
            "name": "ノード名",
            "handler": "ハンドラー名",
            "next": ["次のノード名"],
            "type": "llm|condition|parallel"
        }}
    ],
    "start_node": "最初のノード名"
}}"""
            }]
        )
 
        # JSONを解析してワークフローを構築
        workflow_spec = json.loads(response.content[0].text)
        engine = WorkflowEngine(name=f"dynamic_{task_description[:20]}")
 
        for node_spec in workflow_spec["nodes"]:
            handler_name = node_spec["handler"]
            handler = self.available_handlers.get(handler_name, {}).get("handler")
 
            if handler:
                engine.add_node(WorkflowNode(
                    name=node_spec["name"],
                    type=NodeType(node_spec.get("type", "llm")),
                    handler=handler,
                    next_nodes=node_spec.get("next", [])
                ))
 
        return engine, workflow_spec["start_node"]
 
# 使用例
builder = DynamicWorkflowBuilder(client)
builder.register_handler(
    "extract_text", extract_text_from_pdf,
    "PDFからテキストを抽出"
)
builder.register_handler(
    "translate", translate_text,
    "テキストを翻訳"
)
builder.register_handler(
    "summarize", summarize_text,
    "テキストを要約"
)
 
workflow, start = builder.build_workflow(
    "英語のPDFを読み取り、日本語に翻訳して要約する"
)
result = workflow.run(start, {"pdf_path": "document.pdf"})

# LangGraphによるワークフローエージェント
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal, Annotated
from operator import add
 
# 状態の型定義
class AgentState(TypedDict):
    messages: list
    current_step: str
    classification: str
    draft: str
    review_result: str
    review_count: Annotated[int, "レビュー回数"]
    final_output: str
    token_usage: Annotated[list[dict], add]
 
# ノード関数
def classify_request(state: AgentState) -> AgentState:
    """リクエストを分類する"""
    messages = state["messages"]
    result = llm.invoke(
        f"以下のリクエストを 'simple' / 'complex' に分類:\n{messages[-1]}"
    )
    return {
        "classification": result.content.strip(),
        "token_usage": [{"node": "classify", "tokens": result.usage_metadata}]
    }
 
def handle_simple(state: AgentState) -> AgentState:
    """簡単なリクエストを処理"""
    draft = llm.invoke(f"簡潔に回答: {state['messages'][-1]}")
    return {
        "draft": draft.content,
        "token_usage": [{"node": "handle_simple", "tokens": draft.usage_metadata}]
    }
 
def handle_complex(state: AgentState) -> AgentState:
    """複雑なリクエストを処理"""
    context = ""
    if state.get("review_result") and "FAIL" in state["review_result"]:
        context = f"\n\n前回のフィードバック: {state['review_result']}"
 
    draft = llm.invoke(
        f"詳細に回答: {state['messages'][-1]}{context}"
    )
    return {
        "draft": draft.content,
        "token_usage": [{"node": "handle_complex", "tokens": draft.usage_metadata}]
    }
 
def review(state: AgentState) -> AgentState:
    """回答をレビュー"""
    result = llm.invoke(
        f"回答の品質を評価。PASS/FAIL（FAILの場合は理由も記載）:\n{state['draft']}"
    )
    return {
        "review_result": result.content.strip(),
        "review_count": state.get("review_count", 0) + 1,
        "token_usage": [{"node": "review", "tokens": result.usage_metadata}]
    }
 
def finalize(state: AgentState) -> AgentState:
    """最終出力を生成"""
    return {"final_output": state["draft"]}
 
# 条件分岐関数
def route_by_classification(state: AgentState) -> Literal["simple", "complex"]:
    return "simple" if "simple" in state["classification"] else "complex"
 
def route_by_review(state: AgentState) -> Literal["revise", "finalize"]:
    # 最大3回のレビューで打ち切り
    if state.get("review_count", 0) >= 3:
        return "finalize"
    return "finalize" if "PASS" in state["review_result"] else "revise"
 
# グラフ構築
workflow = StateGraph(AgentState)
 
# ノード追加
workflow.add_node("classify", classify_request)
workflow.add_node("handle_simple", handle_simple)
workflow.add_node("handle_complex", handle_complex)
workflow.add_node("review", review)
workflow.add_node("finalize", finalize)
 
# エッジ追加
workflow.set_entry_point("classify")
workflow.add_conditional_edges("classify", route_by_classification, {
    "simple": "handle_simple",
    "complex": "handle_complex"
})
workflow.add_edge("handle_simple", "review")
workflow.add_edge("handle_complex", "review")
workflow.add_conditional_edges("review", route_by_review, {
    "finalize": "finalize",
    "revise": "handle_complex"  # やり直し
})
workflow.add_edge("finalize", END)
 
# コンパイル・実行
app = workflow.compile()
result = app.invoke({
    "messages": ["マイクロサービスのベストプラクティスを教えて"],
    "current_step": "", "classification": "", "draft": "",
    "review_result": "", "review_count": 0,
    "final_output": "", "token_usage": []
})

# チェックポイント機能付きワークフロー
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
import sqlite3
 
# SQLiteチェックポイント（開発用）
def create_workflow_with_checkpoint():
    """チェックポイント機能付きワークフロー"""
    conn = sqlite3.connect("workflow_checkpoints.db")
    memory = SqliteSaver(conn)
 
    workflow = StateGraph(AgentState)
    # ... ノードとエッジを追加 ...
 
    # チェックポイント付きでコンパイル
    app = workflow.compile(checkpointer=memory)
 
    # スレッドIDで実行状態を管理
    config = {"configurable": {"thread_id": "support-001"}}
 
    # 実行（途中で失敗しても再開可能）
    try:
        result = app.invoke(initial_state, config)
    except Exception as e:
        print(f"エラー発生: {e}")
        # 最後のチェックポイントから状態を取得
        state = app.get_state(config)
        print(f"最後の成功ノード: {state.values}")
 
        # 状態を修正して再開
        app.update_state(config, {"draft": "修正済みドラフト"})
        result = app.invoke(None, config)  # Noneで前回の状態から再開
 
    return result
 
# PostgreSQLチェックポイント（本番用）
def create_production_workflow():
    """本番環境向けチェックポイント"""
    from psycopg_pool import ConnectionPool
 
    pool = ConnectionPool(
        "postgresql://user:pass@localhost/workflows",
        max_size=20
    )
    memory = PostgresSaver(pool)
    memory.setup()  # テーブル作成
 
    workflow = StateGraph(AgentState)
    # ... ノード追加 ...
    app = workflow.compile(checkpointer=memory)
 
    return app

# ストリーミング実行
async def stream_workflow():
    """ワークフローの進捗をリアルタイムで取得"""
    app = create_workflow()
 
    # ノードごとの出力をストリーミング
    async for event in app.astream(
        initial_state,
        config={"configurable": {"thread_id": "stream-001"}}
    ):
        for node_name, output in event.items():
            print(f"[{node_name}] 完了:")
            print(f"  出力: {json.dumps(output, ensure_ascii=False, indent=2)}")
 
    # トークン単位のストリーミング
    async for event in app.astream_events(
        initial_state,
        config={"configurable": {"thread_id": "stream-002"}},
        version="v2"
    ):
        if event["event"] == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            print(chunk.content, end="", flush=True)
        elif event["event"] == "on_chain_end":
            print(f"\n--- {event['name']} 完了 ---")

# 並列ノードの実装
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import NamedTuple
 
class ParallelResult(NamedTuple):
    node_name: str
    result: Any
    duration: float
    success: bool
    error: str = ""
 
class ParallelWorkflow:
    def __init__(self, max_workers: int = 5):
        self.max_workers = max_workers
 
    async def run_parallel_nodes(
        self,
        nodes: list[WorkflowNode],
        state: dict,
        timeout: float = 60.0
    ) -> list[ParallelResult]:
        """複数のノードを並列に実行（タイムアウト付き）"""
 
        async def execute_with_tracking(node: WorkflowNode) -> ParallelResult:
            start = time.time()
            try:
                result = await asyncio.wait_for(
                    asyncio.to_thread(node.handler, state.copy()),
                    timeout=node.timeout
                )
                return ParallelResult(
                    node_name=node.name,
                    result=result,
                    duration=time.time() - start,
                    success=True
                )
            except asyncio.TimeoutError:
                return ParallelResult(
                    node_name=node.name,
                    result=None,
                    duration=time.time() - start,
                    success=False,
                    error=f"タイムアウト ({node.timeout}s)"
                )
            except Exception as e:
                return ParallelResult(
                    node_name=node.name,
                    result=None,
                    duration=time.time() - start,
                    success=False,
                    error=str(e)
                )
 
        tasks = [execute_with_tracking(node) for node in nodes]
        results = await asyncio.gather(*tasks)
 
        # 結果をステートに統合
        for pr in results:
            if pr.success:
                state[f"{pr.node_name}_result"] = pr.result
            else:
                state[f"{pr.node_name}_error"] = pr.error
 
        return list(results)
 
# 使用例: 複数情報源からの並列収集
async def parallel_research(state: dict) -> dict:
    """複数ソースから並列にデータ収集"""
    query = state["query"]
 
    async def search_academic(q: str) -> list[dict]:
        # Semantic Scholar APIなど
        await asyncio.sleep(1)  # シミュレーション
        return [{"title": "Paper A", "source": "academic"}]
 
    async def search_news(q: str) -> list[dict]:
        await asyncio.sleep(0.5)
        return [{"title": "News B", "source": "news"}]
 
    async def search_github(q: str) -> list[dict]:
        await asyncio.sleep(0.8)
        return [{"title": "Repo C", "source": "github"}]
 
    papers, news, repos = await asyncio.gather(
        search_academic(query),
        search_news(query),
        search_github(query)
    )
 
    return {
        "papers": papers,
        "news": news,
        "repos": repos,
        "total_sources": len(papers) + len(news) + len(repos)
    }

# Fan-Out / Fan-In: 入力を分割して並列処理し、結果を集約
class FanOutFanInWorkflow:
    """大量データを分割して並列処理するワークフロー"""
 
    def __init__(self, client: anthropic.Anthropic, max_concurrent: int = 5):
        self.client = client
        self.semaphore = asyncio.Semaphore(max_concurrent)
 
    async def fan_out(self, items: list, chunk_size: int = 10) -> list[list]:
        """入力をチャンクに分割"""
        return [items[i:i+chunk_size] for i in range(0, len(items), chunk_size)]
 
    async def process_chunk(self, chunk: list, prompt_template: str) -> list[dict]:
        """1チャンクをLLMで処理"""
        async with self.semaphore:
            items_text = "\n".join(str(item) for item in chunk)
            response = self.client.messages.create(
                model="claude-haiku-3-20240307",
                max_tokens=1000,
                messages=[{
                    "role": "user",
                    "content": prompt_template.format(items=items_text)
                }]
            )
            return {"chunk_result": response.content[0].text}
 
    async def fan_in(self, results: list[dict]) -> dict:
        """並列処理の結果を集約"""
        all_results = []
        for r in results:
            all_results.append(r["chunk_result"])
 
        # 集約結果をLLMで統合
        combined = "\n---\n".join(all_results)
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{
                "role": "user",
                "content": f"以下の分析結果を統合して総合レポートを作成:\n{combined}"
            }]
        )
        return {"summary": response.content[0].text}
 
    async def run(self, items: list, prompt_template: str) -> dict:
        """Fan-Out → 並列処理 → Fan-In の完全フロー"""
        # Fan-Out
        chunks = await self.fan_out(items)
        print(f"  {len(chunks)}チャンクに分割")
 
        # 並列処理
        tasks = [self.process_chunk(chunk, prompt_template) for chunk in chunks]
        results = await asyncio.gather(*tasks)
        print(f"  {len(results)}チャンク処理完了")
 
        # Fan-In
        summary = await self.fan_in(list(results))
        return summary
 
# 使用例: 1000件のレビューを並列分析
async def analyze_reviews():
    workflow = FanOutFanInWorkflow(client, max_concurrent=10)
 
    reviews = load_reviews()  # 1000件のレビュー
    result = await workflow.run(
        items=reviews,
        prompt_template="以下のレビューを感情分析（positive/negative/neutral）:\n{items}"
    )
    print(result["summary"])

# LangGraphでのMap-Reduce実装
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from operator import add
 
class MapReduceState(TypedDict):
    documents: list[str]
    summaries: Annotated[list[str], add]  # addで結果をマージ
    final_summary: str
 
def map_summarize(state: MapReduceState) -> MapReduceState:
    """各ドキュメントを個別に要約（Mapフェーズ）"""
    summaries = []
    for doc in state["documents"]:
        response = llm.invoke(f"以下を100文字で要約:\n{doc}")
        summaries.append(response.content)
    return {"summaries": summaries}
 
def reduce_combine(state: MapReduceState) -> MapReduceState:
    """要約を統合（Reduceフェーズ）"""
    all_summaries = "\n".join(
        f"{i+1}. {s}" for i, s in enumerate(state["summaries"])
    )
    response = llm.invoke(
        f"以下の要約を統合して総合的なサマリーを作成:\n{all_summaries}"
    )
    return {"final_summary": response.content}
 
# グラフ構築
map_reduce = StateGraph(MapReduceState)
map_reduce.add_node("map", map_summarize)
map_reduce.add_node("reduce", reduce_combine)
map_reduce.set_entry_point("map")
map_reduce.add_edge("map", "reduce")
map_reduce.add_edge("reduce", END)
 
app = map_reduce.compile()

# Pydanticを使った型安全な状態管理
from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime
from enum import Enum
 
class WorkflowPhase(Enum):
    INTAKE = "intake"
    PROCESSING = "processing"
    REVIEW = "review"
    OUTPUT = "output"
    COMPLETED = "completed"
    FAILED = "failed"
 
class WorkflowState(BaseModel):
    """ワークフローの状態（型安全）"""
    # 基本情報
    workflow_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    user_input: str
    step: int = 0
    phase: WorkflowPhase = WorkflowPhase.INTAKE
 
    # 処理結果
    classification: Optional[str] = None
    intermediate_results: list[str] = Field(default_factory=list)
    final_output: Optional[str] = None
 
    # エラー管理
    errors: list[str] = Field(default_factory=list)
    retry_counts: dict[str, int] = Field(default_factory=dict)
 
    # メタデータ
    metadata: dict = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)
 
    # トークン使用量追跡
    total_input_tokens: int = 0
    total_output_tokens: int = 0
 
    @validator("step")
    def step_must_be_non_negative(cls, v):
        if v < 0:
            raise ValueError("step must be non-negative")
        return v
 
    def advance(self) -> "WorkflowState":
        """ステップを進める"""
        return self.model_copy(update={
            "step": self.step + 1,
            "updated_at": datetime.now()
        })
 
    def add_result(self, result: str) -> "WorkflowState":
        """中間結果を追加"""
        new_results = self.intermediate_results + [result]
        return self.model_copy(update={
            "intermediate_results": new_results,
            "updated_at": datetime.now()
        })
 
    def add_error(self, error: str, node_name: str = "") -> "WorkflowState":
        """エラーを記録"""
        new_errors = self.errors + [f"[{node_name}] {error}"]
        retry = self.retry_counts.copy()
        if node_name:
            retry[node_name] = retry.get(node_name, 0) + 1
        return self.model_copy(update={
            "errors": new_errors,
            "retry_counts": retry,
            "updated_at": datetime.now()
        })
 
    def track_tokens(self, input_tokens: int, output_tokens: int) -> "WorkflowState":
        """トークン使用量を追跡"""
        return self.model_copy(update={
            "total_input_tokens": self.total_input_tokens + input_tokens,
            "total_output_tokens": self.total_output_tokens + output_tokens,
        })
 
    def transition_to(self, phase: WorkflowPhase) -> "WorkflowState":
        """フェーズを遷移"""
        return self.model_copy(update={
            "phase": phase,
            "updated_at": datetime.now()
        })
 
    @property
    def estimated_cost(self) -> float:
        """推定コスト（USD）- Claude Sonnet基準"""
        input_cost = self.total_input_tokens * 3.0 / 1_000_000
        output_cost = self.total_output_tokens * 15.0 / 1_000_000
        return input_cost + output_cost

# Redis/SQLiteバックエンドの状態ストア
import json
import sqlite3
from abc import ABC, abstractmethod
 
class StateStore(ABC):
    """状態ストアの抽象基底クラス"""
 
    @abstractmethod
    def save(self, workflow_id: str, state: WorkflowState) -> None:
        pass
 
    @abstractmethod
    def load(self, workflow_id: str) -> Optional[WorkflowState]:
        pass
 
    @abstractmethod
    def list_workflows(self, status: Optional[str] = None) -> list[str]:
        pass
 
class SQLiteStateStore(StateStore):
    """SQLiteベースの状態ストア"""
 
    def __init__(self, db_path: str = "workflow_states.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS workflow_states (
                workflow_id TEXT PRIMARY KEY,
                state_json TEXT NOT NULL,
                phase TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )
        """)
        self.conn.commit()
 
    def save(self, workflow_id: str, state: WorkflowState) -> None:
        self.conn.execute(
            """INSERT OR REPLACE INTO workflow_states
               (workflow_id, state_json, phase, created_at, updated_at)
               VALUES (?, ?, ?, ?, ?)""",
            (
                workflow_id,
                state.model_dump_json(),
                state.phase.value,
                state.created_at.isoformat(),
                state.updated_at.isoformat()
            )
        )
        self.conn.commit()
 
    def load(self, workflow_id: str) -> Optional[WorkflowState]:
        row = self.conn.execute(
            "SELECT state_json FROM workflow_states WHERE workflow_id = ?",
            (workflow_id,)
        ).fetchone()
        if row:
            return WorkflowState.model_validate_json(row[0])
        return None
 
    def list_workflows(self, status: Optional[str] = None) -> list[str]:
        if status:
            rows = self.conn.execute(
                "SELECT workflow_id FROM workflow_states WHERE phase = ?",
                (status,)
            ).fetchall()
        else:
            rows = self.conn.execute(
                "SELECT workflow_id FROM workflow_states"
            ).fetchall()
        return [row[0] for row in rows]
 
class RedisStateStore(StateStore):
    """Redisベースの状態ストア（高速アクセス）"""
 
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        import redis
        self.redis = redis.from_url(redis_url)
        self.prefix = "workflow:"
 
    def save(self, workflow_id: str, state: WorkflowState) -> None:
        key = f"{self.prefix}{workflow_id}"
        self.redis.set(key, state.model_dump_json())
        self.redis.sadd(f"{self.prefix}index:{state.phase.value}", workflow_id)
        # TTL: 30日
        self.redis.expire(key, 30 * 24 * 3600)
 
    def load(self, workflow_id: str) -> Optional[WorkflowState]:
        data = self.redis.get(f"{self.prefix}{workflow_id}")
        if data:
            return WorkflowState.model_validate_json(data)
        return None
 
    def list_workflows(self, status: Optional[str] = None) -> list[str]:
        if status:
            return [
                m.decode()
                for m in self.redis.smembers(f"{self.prefix}index:{status}")
            ]
        # 全件取得
        keys = self.redis.keys(f"{self.prefix}*")
        return [
            k.decode().replace(self.prefix, "")
            for k in keys
            if b"index:" not in k
        ]

# 堅牢なエラーハンドリング付きワークフロー
from dataclasses import dataclass
from typing import Optional
import traceback
 
@dataclass
class NodeError:
    node_name: str
    error_type: str
    message: str
    traceback: str
    retry_attempt: int
    timestamp: float
 
class ResilientWorkflowEngine(WorkflowEngine):
    """障害耐性のあるワークフローエンジン"""
 
    def __init__(self, name: str, state_store: Optional[StateStore] = None):
        super().__init__(name)
        self.state_store = state_store
        self.error_handlers: dict[str, Callable] = {}
        self.fallback_handlers: dict[str, Callable] = {}
        self.errors: list[NodeError] = []
 
    def set_error_handler(self, node_name: str, handler: Callable):
        """ノード固有のエラーハンドラーを設定"""
        self.error_handlers[node_name] = handler
 
    def set_fallback(self, node_name: str, fallback: Callable):
        """フォールバック処理を設定"""
        self.fallback_handlers[node_name] = fallback
 
    def _execute_node_resilient(self, node: WorkflowNode) -> Any:
        """耐障害性のあるノード実行"""
        for attempt in range(node.retry_count + 1):
            try:
                return node.handler(self.state)
            except Exception as e:
                error = NodeError(
                    node_name=node.name,
                    error_type=type(e).__name__,
                    message=str(e),
                    traceback=traceback.format_exc(),
                    retry_attempt=attempt,
                    timestamp=time.time()
                )
                self.errors.append(error)
 
                # ノード固有のエラーハンドラー
                if node.name in self.error_handlers:
                    should_retry = self.error_handlersnode.name
                    if not should_retry:
                        break
 
                if attempt < node.retry_count:
                    wait = min(2 ** attempt * 1.0, 30.0)  # 最大30秒
                    logger.warning(
                        f"リトライ {attempt+1}/{node.retry_count}: "
                        f"{node.name} ({wait:.1f}s待機)"
                    )
                    time.sleep(wait)
 
        # 全リトライ失敗 → フォールバック
        if node.name in self.fallback_handlers:
            logger.info(f"フォールバック実行: {node.name}")
            return self.fallback_handlersnode.name
 
        raise RuntimeError(
            f"ノード '{node.name}' が {node.retry_count + 1}回の試行後に失敗"
        )
 
    def run(self, start_node: str, initial_state: dict) -> dict:
        """チェックポイント付き実行"""
        self.state = initial_state
        workflow_id = self.state.get("workflow_id", str(time.time()))
        current = start_node
 
        while current:
            node = self.nodes[current]
 
            # チェックポイント保存
            if self.state_store:
                ws = WorkflowState(
                    workflow_id=workflow_id,
                    user_input=str(initial_state),
                    step=len(self.execution_log),
                    phase=WorkflowPhase.PROCESSING
                )
                self.state_store.save(workflow_id, ws)
 
            try:
                if node.type == NodeType.CONDITION:
                    branch = node.condition(self.state)
                    current = branch
                else:
                    result = self._execute_node_resilient(node)
                    self.state[f"{node.name}_result"] = result
                    current = node.next_nodes[0] if node.next_nodes else None
            except Exception as e:
                self.state["workflow_error"] = str(e)
                self.state["failed_node"] = node.name
                logger.error(f"ワークフロー停止: {node.name} - {e}")
                break
 
        return self.state
 
# 使用例
engine = ResilientWorkflowEngine("resilient_support", SQLiteStateStore())
 
# エラーハンドラー設定
def handle_llm_error(error: NodeError, state: dict) -> bool:
    """LLMエラーのハンドリング"""
    if "rate_limit" in error.message.lower():
        time.sleep(60)  # レート制限時は60秒待機
        return True  # リトライする
    if "overloaded" in error.message.lower():
        return True  # リトライする
    return False  # リトライしない
 
engine.set_error_handler("handle_technical", handle_llm_error)
 
# フォールバック設定
engine.set_fallback(
    "handle_technical",
    lambda state: "申し訳ございません。技術チームに転送いたします。"
)

# ワークフローノード用サーキットブレーカー
from enum import Enum
import threading
 
class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 遮断中
    HALF_OPEN = "half_open"  # 試行中
 
class CircuitBreaker:
    """ワークフローノード用サーキットブレーカー"""
 
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        success_threshold: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
 
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0.0
        self.lock = threading.Lock()
 
    def can_execute(self) -> bool:
        with self.lock:
            if self.state == CircuitState.CLOSED:
                return True
            elif self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                    return True
                return False
            else:  # HALF_OPEN
                return True
 
    def record_success(self):
        with self.lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
            self.failure_count = 0
 
    def record_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                logger.warning(
                    f"サーキットブレーカー OPEN: "
                    f"{self.failure_count}回連続失敗"
                )
 
def with_circuit_breaker(breaker: CircuitBreaker):
    """サーキットブレーカーデコレータ"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            if not breaker.can_execute():
                raise RuntimeError("サーキットブレーカーが開いています")
            try:
                result = func(*args, **kwargs)
                breaker.record_success()
                return result
            except Exception as e:
                breaker.record_failure()
                raise
        return wrapper
    return decorator
 
# 使用例
llm_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=120)
 
@with_circuit_breaker(llm_breaker)
def call_llm_with_breaker(state: dict) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{"role": "user", "content": state["prompt"]}]
    )
    return response.content[0].text

# 再利用可能なサブワークフロー
class SubWorkflow:
    """独立してテスト・再利用できるサブワークフロー"""
 
    def __init__(self, name: str):
        self.name = name
        self.engine = WorkflowEngine(name=name)
        self.input_schema: dict = {}
        self.output_schema: dict = {}
 
    def define_interface(self, input_keys: list[str], output_keys: list[str]):
        """入出力インターフェースを定義"""
        self.input_schema = {k: True for k in input_keys}
        self.output_schema = {k: True for k in output_keys}
 
    def validate_input(self, state: dict) -> bool:
        """入力の検証"""
        missing = [k for k in self.input_schema if k not in state]
        if missing:
            raise ValueError(f"必須入力が不足: {missing}")
        return True
 
    def execute(self, input_state: dict, start_node: str) -> dict:
        """サブワークフローを実行"""
        self.validate_input(input_state)
        result = self.engine.run(start_node, input_state)
 
        # 出力のみを返す
        output = {k: result.get(k) for k in self.output_schema}
        return output
 
# サブワークフロー: テキスト品質チェック
def build_quality_check_subworkflow() -> SubWorkflow:
    """テキスト品質チェックのサブワークフロー"""
    sub = SubWorkflow("quality_check")
    sub.define_interface(
        input_keys=["text", "criteria"],
        output_keys=["quality_score", "issues", "improved_text"]
    )
 
    # 文法チェックノード
    sub.engine.add_node(WorkflowNode(
        name="grammar_check",
        type=NodeType.LLM,
        handler=lambda s: check_grammar(s["text"]),
        next_nodes=["style_check"]
    ))
 
    # スタイルチェックノード
    sub.engine.add_node(WorkflowNode(
        name="style_check",
        type=NodeType.LLM,
        handler=lambda s: check_style(s["text"], s["criteria"]),
        next_nodes=["score"]
    ))
 
    # スコアリングノード
    sub.engine.add_node(WorkflowNode(
        name="score",
        type=NodeType.LLM,
        handler=lambda s: calculate_score(s),
        next_nodes=[]
    ))
 
    return sub
 
# メインワークフローからサブワークフローを呼び出し
quality_checker = build_quality_check_subworkflow()
 
def content_pipeline():
    engine = WorkflowEngine("content_pipeline")
 
    # 記事生成ノード
    engine.add_node(WorkflowNode(
        name="generate",
        type=NodeType.LLM,
        handler=generate_article,
        next_nodes=["quality_check"]
    ))
 
    # サブワークフロー呼び出しノード
    engine.add_node(WorkflowNode(
        name="quality_check",
        type=NodeType.SUBWORKFLOW,
        handler=lambda state: quality_checker.execute(
            {"text": state["generate_result"], "criteria": "blog"},
            start_node="grammar_check"
        ),
        next_nodes=["publish"]
    ))
 
    # 公開ノード
    engine.add_node(WorkflowNode(
        name="publish",
        type=NodeType.TOOL,
        handler=publish_article,
        next_nodes=[]
    ))
 
    return engine

# 自動コードレビューワークフロー
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
 
class CodeReviewState(TypedDict):
    code: str
    language: str
    diff: str
    static_analysis: dict
    security_issues: list[dict]
    performance_issues: list[dict]
    style_issues: list[dict]
    review_summary: str
    approval_status: str  # "approved", "changes_requested", "blocked"
 
def detect_language(state: CodeReviewState) -> CodeReviewState:
    """プログラミング言語を検出"""
    response = client.messages.create(
        model="claude-haiku-3-20240307",
        max_tokens=20,
        messages=[{
            "role": "user",
            "content": f"このコードの言語を1単語で: {state['code'][:500]}"
        }]
    )
    return {"language": response.content[0].text.strip().lower()}
 
def run_static_analysis(state: CodeReviewState) -> CodeReviewState:
    """静的解析を実行"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": f"""以下の{state['language']}コードを静的解析してください。
JSON形式で結果を返してください:
{{"complexity": "low/medium/high", "issues": [...], "metrics": {{}}}}
 
コード:
{state['code']}"""
        }]
    )
    return {"static_analysis": json.loads(response.content[0].text)}
 
def check_security(state: CodeReviewState) -> CodeReviewState:
    """セキュリティチェック"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        system="セキュリティ専門のコードレビュアーとして脆弱性を分析してください。",
        messages=[{
            "role": "user",
            "content": f"""以下のコードのセキュリティ脆弱性を検出:
- SQLインジェクション
- XSS
- 認証/認可の不備
- 機密情報のハードコード
 
JSON配列で返してください。
 
コード:
{state['code']}"""
        }]
    )
    return {"security_issues": json.loads(response.content[0].text)}
 
def check_performance(state: CodeReviewState) -> CodeReviewState:
    """パフォーマンスチェック"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""パフォーマンス観点でコードレビュー:
- N+1クエリ
- メモリリーク
- 不要なループ
- キャッシュ未使用
 
JSON配列で: {state['code']}"""
        }]
    )
    return {"performance_issues": json.loads(response.content[0].text)}
 
def generate_review_summary(state: CodeReviewState) -> CodeReviewState:
    """レビューサマリーを生成"""
    total_issues = (
        len(state.get("security_issues", []))
        + len(state.get("performance_issues", []))
        + len(state.get("style_issues", []))
    )
 
    has_critical = any(
        issue.get("severity") == "critical"
        for issue in state.get("security_issues", [])
    )
 
    if has_critical:
        status = "blocked"
    elif total_issues > 5:
        status = "changes_requested"
    else:
        status = "approved"
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""以下のレビュー結果を要約:
セキュリティ: {json.dumps(state.get('security_issues', []), ensure_ascii=False)}
パフォーマンス: {json.dumps(state.get('performance_issues', []), ensure_ascii=False)}
ステータス: {status}
 
Markdown形式でレビューサマリーを作成:"""
        }]
    )
 
    return {
        "review_summary": response.content[0].text,
        "approval_status": status
    }
 
# グラフ構築
review_flow = StateGraph(CodeReviewState)
 
review_flow.add_node("detect_lang", detect_language)
review_flow.add_node("static_analysis", run_static_analysis)
review_flow.add_node("security_check", check_security)
review_flow.add_node("performance_check", check_performance)
review_flow.add_node("summarize", generate_review_summary)
 
review_flow.set_entry_point("detect_lang")
review_flow.add_edge("detect_lang", "static_analysis")
# 静的解析後、セキュリティとパフォーマンスを並列実行
review_flow.add_edge("static_analysis", "security_check")
review_flow.add_edge("static_analysis", "performance_check")
review_flow.add_edge("security_check", "summarize")
review_flow.add_edge("performance_check", "summarize")
review_flow.add_edge("summarize", END)
 
code_reviewer = review_flow.compile()

# ブログ記事自動生成ワークフロー
class ContentState(TypedDict):
    topic: str
    target_audience: str
    outline: str
    draft: str
    seo_keywords: list[str]
    review_feedback: str
    review_pass: bool
    final_article: str
    metadata: dict
 
def research_topic(state: ContentState) -> ContentState:
    """トピックをリサーチ"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": f"""以下のトピックについてリサーチして、
記事のアウトラインを作成してください。
 
トピック: {state['topic']}
対象読者: {state['target_audience']}
 
アウトライン形式:
1. 導入部
2. 本文（3-5セクション）
3. まとめ
各セクションに含めるべきポイントを箇条書きで。"""
        }]
    )
    return {"outline": response.content[0].text}
 
def extract_seo_keywords(state: ContentState) -> ContentState:
    """SEOキーワードを抽出"""
    response = client.messages.create(
        model="claude-haiku-3-20240307",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"トピック「{state['topic']}」のSEOキーワードを10個、JSON配列で:"
        }]
    )
    return {"seo_keywords": json.loads(response.content[0].text)}
 
def write_draft(state: ContentState) -> ContentState:
    """記事のドラフトを執筆"""
    feedback_context = ""
    if state.get("review_feedback"):
        feedback_context = f"\n\n前回のフィードバック:\n{state['review_feedback']}"
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        messages=[{
            "role": "user",
            "content": f"""以下のアウトラインに従って記事を執筆してください。
 
アウトライン:
{state['outline']}
 
SEOキーワード（自然に組み込む）:
{', '.join(state.get('seo_keywords', []))}
{feedback_context}
 
Markdown形式で2000-3000文字程度の記事を書いてください。"""
        }]
    )
    return {"draft": response.content[0].text}
 
def review_article(state: ContentState) -> ContentState:
    """記事をレビュー"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""以下の記事を編集者としてレビューしてください。
 
記事:
{state['draft']}
 
評価基準:
1. 正確性
2. 読みやすさ
3. SEO最適化
4. 対象読者への適切さ
 
最初の行に PASS または FAIL を記載し、
FAILの場合は具体的な改善点を列挙してください。"""
        }]
    )
    result = response.content[0].text
    is_pass = result.strip().startswith("PASS")
    return {
        "review_feedback": result,
        "review_pass": is_pass
    }
 
def finalize_article(state: ContentState) -> ContentState:
    """記事を最終化"""
    return {
        "final_article": state["draft"],
        "metadata": {
            "topic": state["topic"],
            "keywords": state.get("seo_keywords", []),
            "word_count": len(state["draft"]),
        }
    }
 
def route_review(state: ContentState) -> Literal["revise", "finalize"]:
    return "finalize" if state.get("review_pass", False) else "revise"
 
# グラフ構築
content_flow = StateGraph(ContentState)
content_flow.add_node("research", research_topic)
content_flow.add_node("seo", extract_seo_keywords)
content_flow.add_node("write", write_draft)
content_flow.add_node("review", review_article)
content_flow.add_node("finalize", finalize_article)
 
content_flow.set_entry_point("research")
content_flow.add_edge("research", "seo")
content_flow.add_edge("seo", "write")
content_flow.add_edge("write", "review")
content_flow.add_conditional_edges("review", route_review, {
    "revise": "write",
    "finalize": "finalize"
})
content_flow.add_edge("finalize", END)
 
content_pipeline = content_flow.compile()

# ワークフロー実行のメトリクス収集
from dataclasses import dataclass, field
from collections import defaultdict
import statistics
 
@dataclass
class WorkflowMetrics:
    """ワークフロー実行メトリクスの収集と分析"""
 
    executions: list[dict] = field(default_factory=list)
    node_durations: dict[str, list[float]] = field(
        default_factory=lambda: defaultdict(list)
    )
    node_failures: dict[str, int] = field(
        default_factory=lambda: defaultdict(int)
    )
    total_tokens: dict[str, int] = field(
        default_factory=lambda: defaultdict(int)
    )
 
    def record_execution(self, execution_log: list[NodeExecution]):
        """実行ログからメトリクスを記録"""
        execution_data = {
            "timestamp": time.time(),
            "nodes": [],
            "total_duration": 0,
            "success": True
        }
 
        for ex in execution_log:
            self.node_durations[ex.node_name].append(ex.duration)
            if ex.status == NodeStatus.FAILED:
                self.node_failures[ex.node_name] += 1
                execution_data["success"] = False
 
            execution_data["nodes"].append({
                "name": ex.node_name,
                "duration": ex.duration,
                "status": ex.status.value
            })
            execution_data["total_duration"] += ex.duration
 
        self.executions.append(execution_data)
 
    def get_bottleneck_nodes(self, top_n: int = 3) -> list[dict]:
        """ボトルネックノードを特定"""
        bottlenecks = []
        for node_name, durations in self.node_durations.items():
            if len(durations) >= 3:
                bottlenecks.append({
                    "node": node_name,
                    "avg_duration": statistics.mean(durations),
                    "p95_duration": sorted(durations)[int(len(durations) * 0.95)],
                    "max_duration": max(durations),
                    "failure_rate": (
                        self.node_failures[node_name] / len(durations)
                    )
                })
 
        return sorted(
            bottlenecks,
            key=lambda x: x["avg_duration"],
            reverse=True
        )[:top_n]
 
    def generate_report(self) -> str:
        """メトリクスレポートを生成"""
        total = len(self.executions)
        success = sum(1 for e in self.executions if e["success"])
 
        report = [
            f"=== ワークフローメトリクスレポート ===",
            f"総実行回数: {total}",
            f"成功率: {success/total*100:.1f}%",
            f"",
            f"--- ノード別パフォーマンス ---"
        ]
 
        for b in self.get_bottleneck_nodes(10):
            report.append(
                f"  {b['node']}: "
                f"平均 {b['avg_duration']:.2f}s, "
                f"P95 {b['p95_duration']:.2f}s, "
                f"失敗率 {b['failure_rate']*100:.1f}%"
            )
 
        return "\n".join(report)

# Streamlitベースのワークフロー監視ダッシュボード
import streamlit as st
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime, timedelta
 
def workflow_dashboard(metrics: WorkflowMetrics):
    """ワークフロー監視ダッシュボード"""
    st.title("ワークフローモニタリング")
 
    # KPIカード
    col1, col2, col3, col4 = st.columns(4)
    total = len(metrics.executions)
    success = sum(1 for e in metrics.executions if e["success"])
 
    col1.metric("総実行回数", total)
    col2.metric("成功率", f"{success/max(total,1)*100:.1f}%")
    col3.metric(
        "平均実行時間",
        f"{statistics.mean([e['total_duration'] for e in metrics.executions]):.1f}s"
    )
    col4.metric(
        "アクティブ実行",
        sum(1 for e in metrics.executions
            if time.time() - e["timestamp"] < 60)
    )
 
    # ノード別実行時間のヒートマップ
    st.subheader("ノード別パフォーマンス")
    bottlenecks = metrics.get_bottleneck_nodes(10)
    if bottlenecks:
        fig = px.bar(
            bottlenecks,
            x="node",
            y="avg_duration",
            color="failure_rate",
            color_continuous_scale="RdYlGn_r",
            title="ノード別平均実行時間と失敗率"
        )
        st.plotly_chart(fig)
 
    # 実行時間のトレンド
    st.subheader("実行時間トレンド")
    if metrics.executions:
        times = [e["timestamp"] for e in metrics.executions]
        durations = [e["total_duration"] for e in metrics.executions]
        fig = go.Figure()
        fig.add_trace(go.Scatter(
            x=[datetime.fromtimestamp(t) for t in times],
            y=durations,
            mode="lines+markers",
            name="実行時間"
        ))
        st.plotly_chart(fig)
 
# Streamlitアプリとして実行: streamlit run dashboard.py