feat(examples/python): add runnable example

Author: AgentPatterns

examples/agent-patterns/multi-agent-collaboration/python/README.md

@@ -0,0 +1,61 @@
+# Multi-Agent Collaboration - Python Implementation
+
+Runnable implementation of a multi-agent collaboration flow where specialized
+agents align on one launch decision across multiple rounds.
+
+---
+
+## Quick start
+
+```bash
+# (optional) create venv
+python -m venv .venv && source .venv/bin/activate
+
+# install dependencies
+pip install -r requirements.txt
+
+# set API key
+export OPENAI_API_KEY="sk-..."
+
+# run the agent
+python main.py
+```
+
+## Full walkthrough
+
+Read the complete implementation guide:
+https://agentpatterns.tech/en/agent-patterns/multi-agent-collaboration
+
+## What's inside
+
+- Role-based team loop (`demand`, `finance`, `risk`)
+- Shared context board passed to every role
+- Collaboration gateway with strict contribution contract
+- Round policy: conflict detection + consensus rule
+- Runtime budgets (`max_rounds`, `max_messages`, `max_seconds`)
+- Final synthesis after team alignment
+- Trace and history for auditability
+
+## Project layout
+
+```text
+examples/
+  agent-patterns/
+    multi-agent-collaboration/
+      python/
+        README.md
+        main.py
+        llm.py
+        gateway.py
+        signals.py
+        requirements.txt
+```
+
+## Notes
+
+- Code and README are English-only by design.
+- The website provides multilingual explanations and theory.
+
+## License
+
+MIT

examples/agent-patterns/multi-agent-collaboration/python/gateway.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+class StopRun(Exception):
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
+@dataclass(frozen=True)
+class Budget:
+    max_rounds: int = 3
+    max_messages: int = 12
+    max_seconds: int = 40
+    min_go_votes: int = 2
+
+
+ALLOWED_STANCES = {"go", "caution", "block"}
+
+
+
+def _is_number(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+
+def validate_contribution(raw: Any, *, allowed_agents: set[str]) -> dict[str, Any]:
+    if not isinstance(raw, dict):
+        raise StopRun("invalid_contribution:not_object")
+
+    required = {"agent", "stance", "summary", "confidence", "actions"}
+    if not required.issubset(raw.keys()):
+        raise StopRun("invalid_contribution:missing_keys")
+
+    agent = raw["agent"]
+    stance = raw["stance"]
+    summary = raw["summary"]
+    confidence = raw["confidence"]
+    actions = raw["actions"]
+
+    if not isinstance(agent, str) or not agent.strip():
+        raise StopRun("invalid_contribution:agent")
+    agent = agent.strip()
+    if agent not in allowed_agents:
+        raise StopRun(f"invalid_contribution:agent_not_allowed:{agent}")
+
+    if not isinstance(stance, str) or stance.strip() not in ALLOWED_STANCES:
+        raise StopRun("invalid_contribution:stance")
+    stance = stance.strip()
+
+    if not isinstance(summary, str) or not summary.strip():
+        raise StopRun("invalid_contribution:summary")
+
+    if not _is_number(confidence):
+        raise StopRun("invalid_contribution:confidence_type")
+    confidence = float(confidence)
+    if not (0.0 <= confidence <= 1.0):
+        raise StopRun("invalid_contribution:confidence_range")
+
+    if not isinstance(actions, list) or not actions:
+        raise StopRun("invalid_contribution:actions")
+
+    normalized_actions: list[str] = []
+    for item in actions:
+        if not isinstance(item, str) or not item.strip():
+            raise StopRun("invalid_contribution:action_item")
+        normalized_actions.append(item.strip())
+
+    # Ignore unknown keys to tolerate extra LLM fields.
+    return {
+        "agent": agent,
+        "stance": stance,
+        "summary": summary.strip(),
+        "confidence": round(confidence, 3),
+        "actions": normalized_actions[:3],
+    }
+
+
+
+def detect_conflicts(contributions: list[dict[str, Any]]) -> list[str]:
+    if not contributions:
+        return ["no_contributions"]
+
+    stances = {item["stance"] for item in contributions}
+    conflicts: list[str] = []
+
+    if "go" in stances and "caution" in stances and "block" not in stances:
+        conflicts.append("go_vs_caution")
+    if "block" in stances and len(stances) > 1:
+        conflicts.append("blocking_vs_non_block")
+    if len(stances) == 3:
+        conflicts.append("high_divergence")
+
+    return conflicts
+
+
+
+def decide_round_outcome(
+    contributions: list[dict[str, Any]],
+    *,
+    min_go_votes: int,
+) -> str | None:
+    go_votes = sum(1 for item in contributions if item["stance"] == "go")
+    caution_votes = sum(1 for item in contributions if item["stance"] == "caution")
+    block_votes = sum(1 for item in contributions if item["stance"] == "block")
+
+    if block_votes >= 2:
+        return "no_go"
+
+    if block_votes > 0:
+        return None
+    if go_votes >= min_go_votes and caution_votes == 0:
+        return "go"
+    if go_votes >= min_go_votes and caution_votes > 0:
+        return "go_with_caution"
+    return None
+
+
+class CollaborationGateway:
+    def __init__(self, *, allow: set[str], budget: Budget):
+        self.allow = set(allow)
+        self.budget = budget
+        self.message_count = 0
+
+    def _consume_message_budget(self) -> None:
+        self.message_count += 1
+        if self.message_count > self.budget.max_messages:
+            raise StopRun("max_messages")
+
+    def accept(self, raw: Any, *, expected_agent: str) -> dict[str, Any]:
+        if expected_agent not in self.allow:
+            raise StopRun(f"agent_denied:{expected_agent}")
+
+        self._consume_message_budget()
+        contribution = validate_contribution(raw, allowed_agents=self.allow)
+
+        if contribution["agent"] != expected_agent:
+            raise StopRun(f"invalid_contribution:agent_mismatch:{expected_agent}")
+
+        return contribution

examples/agent-patterns/multi-agent-collaboration/python/llm.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from openai import APIConnectionError, APITimeoutError, OpenAI
+
+MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")
+LLM_TIMEOUT_SECONDS = float(os.getenv("OPENAI_TIMEOUT_SECONDS", "60"))
+
+
+class LLMTimeout(Exception):
+    pass
+
+
+class LLMEmpty(Exception):
+    pass
+
+
+COMMON_RULES = """
+Return exactly one JSON object with this shape:
+{
+  "agent": "<role_name>",
+  "stance": "go|caution|block",
+  "summary": "one short paragraph",
+  "confidence": 0.0,
+  "actions": ["action 1", "action 2"]
+}
+
+Rules:
+- Use only the provided facts.
+- Keep actions concrete and operational.
+- Do not output markdown or extra keys.
+""".strip()
+
+AGENT_PROMPTS = {
+    "demand_analyst": (
+        "You are Demand Analyst. Focus on demand capacity, conversion, and traffic risks. "
+        "Decide whether launch is feasible from growth and operational demand perspective."
+    ),
+    "finance_analyst": (
+        "You are Finance Analyst. Focus on revenue, margin, campaign cost, and downside exposure. "
+        "Decide if launch economics are acceptable."
+    ),
+    "risk_analyst": (
+        "You are Risk Analyst. Focus on payment reliability, chargebacks, and incidents. "
+        "Prioritize safety and compliance risk containment."
+    ),
+    "legal_analyst": (
+        "You are Legal Analyst. Focus on regulatory, compliance, consumer protection, and policy constraints. "
+        "Flag launch blockers and required mitigations."
+    ),
+}
+
+FINAL_SYSTEM_PROMPT = """
+You are a launch readiness editor.
+Write a short operations brief in English.
+Include:
+- final decision (go/go_with_caution/no_go)
+- why the team agreed
+- top 2 immediate actions
+Use only evidence from collaboration history.
+""".strip()
+
+
+
+def _get_client() -> OpenAI:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise EnvironmentError(
+            "OPENAI_API_KEY is not set. Run: export OPENAI_API_KEY='sk-...'"
+        )
+    return OpenAI(api_key=api_key)
+
+
+
+def _round_summaries(history: list[dict[str, Any]], limit: int = 2) -> list[dict[str, Any]]:
+    summaries: list[dict[str, Any]] = []
+    for row in history[-limit:]:
+        summaries.append(
+            {
+                "round": row.get("round"),
+                "decision": row.get("decision"),
+                "conflicts": row.get("conflicts", []),
+                "stances": [
+                    {
+                        "agent": item.get("agent"),
+                        "stance": item.get("stance"),
+                        "confidence": item.get("confidence"),
+                    }
+                    for item in row.get("contributions", [])
+                ],
+            }
+        )
+    return summaries
+
+
+
+def propose_contribution(
+    *,
+    role: str,
+    goal: str,
+    shared_context: dict[str, Any],
+    history: list[dict[str, Any]],
+    open_conflicts: list[str],
+) -> dict[str, Any]:
+    system = AGENT_PROMPTS.get(role)
+    if not system:
+        raise ValueError(f"unknown_role:{role}")
+
+    payload = {
+        "goal": goal,
+        "role": role,
+        "shared_context": shared_context,
+        "recent_rounds": _round_summaries(history, limit=2),
+        "open_conflicts": open_conflicts,
+    }
+
+    client = _get_client()
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL,
+            temperature=0,
+            timeout=LLM_TIMEOUT_SECONDS,
+            response_format={"type": "json_object"},
+            messages=[
+                {"role": "system", "content": f"{system}\n\n{COMMON_RULES}"},
+                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
+            ],
+        )
+    except (APITimeoutError, APIConnectionError) as exc:
+        raise LLMTimeout("llm_timeout") from exc
+
+    text = completion.choices[0].message.content or "{}"
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return {"invalid": True, "raw": text}
+
+
+
+def compose_final_answer(
+    *,
+    goal: str,
+    final_decision: str,
+    history: list[dict[str, Any]],
+) -> str:
+    payload = {
+        "goal": goal,
+        "final_decision": final_decision,
+        "history": history,
+    }
+
+    client = _get_client()
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL,
+            temperature=0,
+            timeout=LLM_TIMEOUT_SECONDS,
+            messages=[
+                {"role": "system", "content": FINAL_SYSTEM_PROMPT},
+                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
+            ],
+        )
+    except (APITimeoutError, APIConnectionError) as exc:
+        raise LLMTimeout("llm_timeout") from exc
+
+    text = (completion.choices[0].message.content or "").strip()
+    if not text:
+        raise LLMEmpty("llm_empty")
+    return text