Add multi-agent orchestration recipe

recipes/multi_agent_orchestration/README.md

@@ -0,0 +1,64 @@
+# Multi-Agent Orchestration with Llama
+
+A recipe for building multi-agent systems using Llama models, with context layers, agent discovery, and safety wrappers.
+
+## What You'll Build
+
+A 3-agent pipeline where:
+1. **Researcher** analyzes a topic and produces structured findings
+2. **Builder** creates a solution based on research
+3. **Reviewer** evaluates the solution against quality criteria
+
+Each agent has a persistent identity, discovers others via capability cards, and every LLM call is wrapped with safety guards.
+
+## Concepts Demonstrated
+
+| Concept | What | Why |
+|---|---|---|
+| **Context Layers** | Agents have identity (static) + state (dynamic) + knowledge (searchable) | Consistent behavior across sessions |
+| **Agent Cards** | JSON capability descriptors per agent | Agents find each other without hardcoding |
+| **Safety Guard** | Output validation, cost caps, rollback | Production-grade reliability |
+| **Phase Gates** | Research → Build → Review with quality checks | Prevent garbage from propagating |
+
+## Quick Start
+
+```bash
+pip install -r requirements.txt
+# Set your Llama API endpoint (or use ollama)
+export LLAMA_BASE_URL=http://localhost:11434/v1
+export LLAMA_MODEL=llama3.2
+
+python orchestrator.py "Build a CLI tool that converts CSV to JSON"
+```
+
+## Files
+
+| File | What |
+|---|---|
+| `context_layers.py` | 4-layer context system (identity → state → relevant → archive) |
+| `agent_cards.py` | Agent capability discovery and routing |
+| `safety_guard.py` | Output validation, cost tracking, rollback |
+| `orchestrator.py` | 3-agent pipeline with phase gates |
+| `requirements.txt` | Dependencies |
+
+## How It Works
+
+```
+User provides task
+    ↓
+Orchestrator reads agent cards → finds Researcher
+    ↓
+Researcher (context: identity + task) → structured findings
+    ↓ [safety guard: validate JSON output]
+Orchestrator finds Builder
+    ↓
+Builder (context: identity + research findings) → solution
+    ↓ [safety guard: validate + cost check]
+Orchestrator finds Reviewer
+    ↓
+Reviewer (context: identity + solution + criteria) → evaluation
+    ↓ [phase gate: score > 7/10 to pass]
+Output: solution + evaluation report
+```
+
+Built by [PA·co](https://github.com/PenguinAlleyApps/paco-framework) — A Penguin Alley System.

recipes/multi_agent_orchestration/agent_cards.py

@@ -0,0 +1,78 @@
+"""
+Agent Cards — Capability discovery for multi-agent routing.
+
+Instead of hardcoding "send this to Agent X", agents publish
+their capabilities as JSON cards. The orchestrator searches
+cards to find the right agent for each task.
+"""
+from dataclasses import dataclass, field
+
+
+@dataclass
+class AgentCard:
+    """JSON-serializable capability descriptor for an agent."""
+    name: str
+    capabilities: list[str]
+    inputs: list[str]
+    outputs: list[str]
+
+    def matches(self, query: str) -> bool:
+        """Check if this agent matches a capability query."""
+        q = query.lower()
+        return any(q in cap.lower() for cap in self.capabilities)
+
+    def to_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "capabilities": self.capabilities,
+            "inputs": self.inputs,
+            "outputs": self.outputs,
+        }
+
+
+class CardDirectory:
+    """Registry of all agent cards. Agents find each other here."""
+
+    def __init__(self):
+        self._cards: dict[str, AgentCard] = {}
+
+    def register(self, card: AgentCard):
+        self._cards[card.name] = card
+
+    def find(self, capability: str) -> list[AgentCard]:
+        """Find agents that match a capability query."""
+        return [c for c in self._cards.values() if c.matches(capability)]
+
+    def get(self, name: str) -> AgentCard | None:
+        return self._cards.get(name)
+
+    def list_all(self) -> list[AgentCard]:
+        return list(self._cards.values())
+
+
+# Pre-built directory for the 3-agent pipeline
+def build_default_directory() -> CardDirectory:
+    directory = CardDirectory()
+
+    directory.register(AgentCard(
+        name="Researcher",
+        capabilities=["research", "analysis", "market_study", "competitive_intel"],
+        inputs=["topic", "question", "domain"],
+        outputs=["structured_findings_json"],
+    ))
+
+    directory.register(AgentCard(
+        name="Builder",
+        capabilities=["build", "implement", "code", "create", "develop"],
+        inputs=["research_findings", "spec", "requirements"],
+        outputs=["code", "implementation_plan"],
+    ))
+
+    directory.register(AgentCard(
+        name="Reviewer",
+        capabilities=["review", "evaluate", "qa", "quality_check"],
+        inputs=["solution", "criteria", "research_context"],
+        outputs=["evaluation_json", "score", "verdict"],
+    ))
+
+    return directory

recipes/multi_agent_orchestration/context_layers.py

@@ -0,0 +1,76 @@
+"""
+Context Layers — Give agents persistent identity + dynamic state.
+
+4 layers, from most stable to most dynamic:
+  Layer 1 (Identity): Who the agent IS. Never changes mid-session.
+  Layer 2 (State): What's happening NOW. Changes every call.
+  Layer 3 (Relevant): Knowledge retrieved for this specific task.
+  Layer 4 (Archive): Everything stored, not loaded unless searched.
+"""
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class AgentContext:
+    """Assembled context for an agent, built from 4 layers."""
+
+    # Layer 1: Identity (static per session)
+    name: str
+    role: str
+    instructions: str
+
+    # Layer 2: State (dynamic per call)
+    current_task: str = ""
+    previous_output: str = ""
+
+    # Layer 3: Relevant (retrieved per task)
+    knowledge: list[str] = field(default_factory=list)
+
+    def to_system_prompt(self) -> str:
+        """Assemble all layers into a system prompt."""
+        parts = [
+            f"# Identity\nYou are {self.name}, a {self.role}.\n{self.instructions}",
+        ]
+        if self.current_task:
+            parts.append(f"\n# Current State\nTask: {self.current_task}")
+        if self.previous_output:
+            parts.append(f"Previous step output:\n{self.previous_output}")
+        if self.knowledge:
+            parts.append(
+                "\n# Relevant Knowledge\n" + "\n".join(f"- {k}" for k in self.knowledge)
+            )
+        return "\n".join(parts)
+
+
+# Pre-built agent identities
+RESEARCHER = AgentContext(
+    name="Researcher",
+    role="research analyst",
+    instructions=(
+        "You analyze topics thoroughly and produce structured findings.\n"
+        "Output JSON with: {summary, key_points[], risks[], opportunities[]}\n"
+        "Be specific. Cite reasoning. Flag uncertainties."
+    ),
+)
+
+BUILDER = AgentContext(
+    name="Builder",
+    role="implementation engineer",
+    instructions=(
+        "You build solutions based on research findings.\n"
+        "Output working code or detailed implementation plan.\n"
+        "Prioritize: correctness > speed > elegance."
+    ),
+)
+
+REVIEWER = AgentContext(
+    name="Reviewer",
+    role="quality reviewer",
+    instructions=(
+        "You evaluate solutions against quality criteria.\n"
+        "Output JSON with: {score (1-10), strengths[], issues[], verdict}\n"
+        "Score >= 7 = PASS. Score < 7 = needs revision.\n"
+        "Be constructive but honest. No rubber stamping."
+    ),
+)