マルチエージェント協調
設計パターンから本番運用まで（2026）

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class PipelineState(TypedDict):
    query: str
    retrieved_docs: str
    analysis: str
    final_report: str

def retrieval_agent(state: PipelineState):
    docs = search_knowledge_base(state["query"])
    return {"retrieved_docs": docs}

def analysis_agent(state: PipelineState):
    result = llm.invoke(f"Analyze the following: {state['retrieved_docs']}")
    return {"analysis": result.content}

def writer_agent(state: PipelineState):
    report = llm.invoke(f"Write a report based on: {state['analysis']}")
    return {"final_report": report.content}

builder = StateGraph(PipelineState)
builder.add_node("retriever", retrieval_agent)
builder.add_node("analyzer", analysis_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "retriever")
builder.add_edge("retriever", "analyzer")
builder.add_edge("analyzer", "writer")
builder.add_edge("writer", END)
pipeline = builder.compile()

from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ResearchState(TypedDict):
    query: str
    research_results: Annotated[list, operator.add]
    final_synthesis: str

def supervisor(state: ResearchState):
    subtasks = [
        {"query": state["query"], "source": "academic"},
        {"query": state["query"], "source": "industry"},
        {"query": state["query"], "source": "news"},
    ]
    return [Send("research_worker", task) for task in subtasks]

def research_worker(state: dict):
    result = search_by_source(state["query"], state["source"])
    return {"research_results": [result]}

def synthesizer(state: ResearchState):
    combined = "\n".join(state["research_results"])
    synthesis = llm.invoke(f"Synthesize the following research: {combined}")
    return {"final_synthesis": synthesis.content}

builder = StateGraph(ResearchState)
builder.add_node("research_worker", research_worker)
builder.add_node("synthesizer", synthesizer)
builder.add_conditional_edges(START, supervisor, ["research_worker"])
builder.add_edge("research_worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()

KEYWORD_ROUTING = {
    "code": "code_agent",
    "search": "search_agent",
    "query": "search_agent",
    "data": "data_agent",
}

def supervisor_with_fast_path(state):
    query = state["query"].lower()
    for keyword, agent_name in KEYWORD_ROUTING.items():
        if keyword in query:
            return {"next": agent_name}
    routing_prompt = f"""
    User request: {state['query']}
    Available agents: code_agent, search_agent, data_agent
    Return only the most suitable agent name.
    """
    decision = llm.invoke(routing_prompt)
    return {"next": decision.content.strip()}

import autogen

reviewer_1 = autogen.AssistantAgent(
    name="SecurityReviewer",
    system_message="You are a security expert focused on vulnerabilities in code."
)
reviewer_2 = autogen.AssistantAgent(
    name="PerformanceReviewer",
    system_message="You are a performance expert focused on efficiency and resource usage."
)
human_proxy = autogen.UserProxyAgent(
    name="CodeAuthor",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=2,
    is_termination_msg=lambda x: "APPROVED" in x.get("content", "")
)
groupchat = autogen.GroupChat(
    agents=[human_proxy, reviewer_1, reviewer_2],
    messages=[],
    max_round=6
)
manager = autogen.GroupChatManager(groupchat=groupchat)

from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("data-agent-mcp")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_customer_db",
            description="Query customer database by id, name, or email",
            inputSchema={
                "type": "object",
                "properties": {
                    "field": {"type": "string", "enum": ["id", "name", "email"]},
                    "value": {"type": "string"}
                },
                "required": ["field", "value"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_customer_db":
        result = db.query(arguments["field"], arguments["value"])
        return [TextContent(type="text", text=str(result))]

import httpx

async def discover_and_delegate(agent_url: str, task: str):
    card_response = await httpx.get(f"{agent_url}/.well-known/agent.json")
    agent_card = card_response.json()
    available_skills = [s["id"] for s in agent_card["skills"]]
    if "web_research" not in available_skills:
        raise ValueError(f"Agent {agent_card['name']} does not support web_research")
    payload = {
        "jsonrpc": "2.0",
        "method": "message/send",
        "id": "task-001",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": task}]
            }
        }
    }
    response = await httpx.post(agent_card["url"], json=payload)
    return response.json()

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt

with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/agentdb") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "user-session-12345"}}
    result = graph.invoke({"query": "Analyze Q2 earnings"}, config)

def high_risk_action_agent(state):
    proposed_action = plan_action(state)
    human_decision = interrupt({
        "proposed_action": proposed_action,
        "risk_level": "HIGH",
        "message": "This operation will modify the production database. Please confirm."
    })
    if human_decision["approved"]:
        return execute_action(proposed_action)
    return {"status": "cancelled", "reason": human_decision.get("reason")}

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "CLOSED"
        self.last_failure_time = None

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if self.state == "OPEN":
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "HALF_OPEN"
                else:
                    raise Exception("Circuit breaker OPEN - agent temporarily unavailable")
            try:
                result = await func(*args, **kwargs)
                if self.state == "HALF_OPEN":
                    self.state = "CLOSED"
                    self.failure_count = 0
                return result
            except Exception:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "OPEN"
                raise
        return wrapper

class TokenBudgetManager:
    def __init__(self, total_budget: int = 100_000):
        self.total_budget = total_budget
        self.used_tokens = 0
        self.agent_usage = {}

    def check_budget(self, agent_name: str, estimated_tokens: int) -> bool:
        remaining = self.total_budget - self.used_tokens
        if estimated_tokens > remaining:
            raise BudgetExceededException(
                f"Agent {agent_name} requested {estimated_tokens} tokens, "
                f"but only {remaining} tokens remain"
            )
        return True

    def record_usage(self, agent_name: str, actual_tokens: int):
        self.used_tokens += actual_tokens
        self.agent_usage[agent_name] = self.agent_usage.get(agent_name, 0) + actual_tokens

from opentelemetry import trace
import uuid

tracer = trace.get_tracer("multi-agent-system")

def traced_agent_call(agent_name: str, task: dict, correlation_id: str = None):
    if not correlation_id:
        correlation_id = str(uuid.uuid4())
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("correlation.id", correlation_id)
        span.set_attribute("task.type", task.get("type", "unknown"))
        try:
            result = agent_registry[agent_name].run(task)
            span.set_attribute("agent.tokens_used", result.get("tokens", 0))
            span.set_attribute("agent.status", "success")
            return result
        except Exception as e:
            span.set_attribute("agent.status", "error")
            span.set_attribute("error.message", str(e))
            raise

MONITORING_METRICS = {
    "task_success_rate": "End-to-end task completion rate (target: >85%)",
    "e2e_latency_p95": "P95 end-to-end latency (target: <30s)",
    "total_cost_per_task": "Average token cost per task",
    "agent_error_rate": "Per-agent error rate (target: <5%)",
    "agent_retry_count": "Retry count (high retries = investigate)",
    "tool_call_budget_usage": "Tool call count vs budget ratio",
    "output_quality_score": "Output quality score",
    "goal_alignment_score": "Goal alignment score",
    "hallucination_rate": "Hallucination detection rate",
}

def evaluate_agent_output(original_task: str, agent_output: str) -> dict:
    evaluation_prompt = f"""
    Original task: {original_task}
    Agent output: {agent_output}
    Score completeness, accuracy, relevance, and hallucination (1-5).
    Return JSON: {{"completeness": x, "accuracy": x, "relevance": x,
    "hallucination_detected": true/false, "comments": "..."}}
    """
    evaluation = llm.invoke(evaluation_prompt)
    return json.loads(evaluation.content)

def validate_agent_output(output: dict, schema: dict) -> bool:
    jsonschema.validate(output, schema)
    if output.get("confidence_score", 1.0) < 0.7:
        raise LowConfidenceError(f"Agent output confidence too low: {output['confidence_score']}")
    required_fields = schema.get("required", [])
    missing = [f for f in required_fields if not output.get(f)]
    if missing:
        raise MissingFieldsError(f"Missing required fields: {missing}")
    return True

MAX_ITERATIONS = 10
MAX_TOOL_CALLS_PER_AGENT = 20
MAX_TOTAL_TOKENS = 50_000

class ProductionGuardrails:
    def validate_input(self, user_input: str) -> str:
        if len(user_input) > 10000:
            raise InputTooLongError("Input exceeds 10000 character limit")
        injection_patterns = ["ignore previous instructions", "forget everything"]
        for pattern in injection_patterns:
            if pattern.lower() in user_input.lower():
                raise PromptInjectionError("Potential prompt injection detected")
        return user_input.strip()

    def validate_output(self, output: str) -> str:
        output = self.pii_filter.redact(output)
        if self.content_classifier.is_harmful(output):
            raise HarmfulContentError("Output contains harmful content")
        return output

builder.add_node("supervisor", supervisor_node, defer=True)

タスクに明確な直列依存ステップがありますか？
├─ はい → サブタスクは並列実行できますか？
│        ├─ いいえ → 【順次パイプライン】
│        └─ はい → 【並列ファンアウト + 順次パイプラインのハイブリッド】
└─ いいえ → 意思決定権を持つ Agent はいますか？
         ├─ はい → サブチームが必要な規模ですか？
         │        ├─ いいえ → 【Supervisor-Worker 階層パターン】
         │        └─ はい → 【階層型（Supervisors of Supervisors）】
         └─ いいえ → 長時間の非同期タスクですか？
                  ├─ はい → 【ブラックボードアーキテクチャ】
                  └─ いいえ → Agent 数は 5 以下ですか？
                           ├─ はい → 【Swarm（終了条件を必ず設定）】
                           └─ いいえ → 【階層パターンへの再分割を検討】