Deep Agents: Real-World Examples

from pathlib import Path
import sys

ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from shipit_agent import Agent, Pipeline, step, parallel
from shipit_agent.deep import (
    GoalAgent,
    Goal,
    ReflectiveAgent,
    Supervisor,
    Channel,
    AgentMessage,
)
from examples.run_multi_tool_agent import build_llm_from_env
from IPython.display import display, Markdown

llm = build_llm_from_env("bedrock")


def print_event(event):
    """Pretty-print an event WITH its output content."""
    ICONS = {
        "run_started": "🚀",
        "run_completed": "✅",
        "planning_started": "📋",
        "planning_completed": "📋",
        "step_started": "▶️",
        "tool_completed": "📦",
        "tool_called": "🔧",
        "tool_failed": "❌",
        "reasoning_started": "🧠",
        "reasoning_completed": "🧠",
    }
    icon = ICONS.get(event.type, "•")
    worker = event.payload.get("worker", "")
    prefix = f"[{worker}] " if worker else ""
    print(f"\n{icon} {prefix}{event.message}")

    output = event.payload.get("output", "")
    if output and event.type in ("tool_completed", "run_completed"):
        print(f"{'─' * 60}")
        print(output[:800])
        if len(output) > 800:
            print(f"... ({len(output)} chars total)")
        print(f"{'─' * 60}")

    subtasks = event.payload.get("subtasks")
    if subtasks:
        for i, t in enumerate(subtasks, 1):
            print(f"  {i}. {t}")

    criteria = event.payload.get("criteria_met")
    if criteria is not None:
        labels = ["✅" if c else "❌" for c in criteria]
        print(f"  Criteria: {' '.join(labels)}")

    quality = event.payload.get("quality")
    if quality is not None:
        bar = "█" * int(quality * 10) + "░" * (10 - int(quality * 10))
        print(f"  Quality: [{bar}] {quality:.2f}")

    feedback = event.payload.get("feedback")
    if feedback:
        print(f"  Feedback: {feedback[:200]}")

# Create a social media content team — each worker has web search & all tools
supervisor = Supervisor.with_builtins(
    llm=llm,
    worker_configs=[
        {
            "name": "trend-researcher",
            "prompt": "You are a social media trend researcher. Search the web for the latest trends. Return concise bullet points.",
            "capabilities": ["web search", "research"],
        },
        {
            "name": "content-writer",
            "prompt": "You are a viral social media content creator. Write engaging posts with emojis and hashtags.",
            "capabilities": ["writing", "social media"],
        },
        {
            "name": "content-reviewer",
            "prompt": "You are a social media strategist. Review content for engagement. Rate 1-10.",
            "capabilities": ["review", "strategy"],
        },
    ],
    max_delegations=5,
)

# Stream with full output visible
print("=== Social Media Content Team ===\n")
for event in supervisor.stream(
    "Create 3 viral social media posts about AI agents for developers"
):
    print_event(event)

# Create a typed communication channel
channel = Channel(name="content-pipeline")

# --- Step 1: Researcher agent gathers info ---
researcher_agent = Agent(
    llm=llm, prompt="You are a researcher. Return key findings as bullet points."
)
research_result = researcher_agent.run(
    "Find 5 key facts about WebAssembly adoption in 2025"
)

# Send findings through the channel with structured data
channel.send(
    AgentMessage(
        from_agent="researcher",
        to_agent="writer",
        type="research_complete",
        data={
            "topic": "WebAssembly adoption",
            "findings": research_result.output,
            "word_count": len(research_result.output.split()),
            "confidence": 0.9,
        },
        requires_ack=True,
    )
)
print(f"Researcher sent findings ({len(research_result.output)} chars)")

# --- Step 2: Writer receives and processes ---
msg = channel.receive(agent="writer")
print(f"\nWriter received '{msg.type}' from {msg.from_agent}")
print(f"  Confidence: {msg.data['confidence']}")
print(f"  Word count: {msg.data['word_count']}")
channel.ack(msg)
print(f"  Acknowledged: {msg.acknowledged}")

# Writer creates content from the research
writer_agent = Agent(
    llm=llm, prompt="You are a technical writer. Write concise, clear content."
)
draft = writer_agent.run(
    f"Write a 2-paragraph summary about WebAssembly using these findings:\n{msg.data['findings']}"
)

# Send draft to reviewer
channel.send(
    AgentMessage(
        from_agent="writer",
        to_agent="reviewer",
        type="draft_ready",
        data={"draft": draft.output, "based_on": msg.type},
    )
)
print(f"\nWriter sent draft to reviewer ({len(draft.output)} chars)")

# --- Step 3: Reviewer receives and reviews ---
review_msg = channel.receive(agent="reviewer")
reviewer_agent = Agent(
    llm=llm,
    prompt="You are a critical editor. Rate quality 1-10 and suggest improvements.",
)
review = reviewer_agent.run(f"Review this draft:\n{review_msg.data['draft']}")

channel.send(
    AgentMessage(
        from_agent="reviewer",
        to_agent="writer",
        type="review_complete",
        data={"review": review.output, "status": "approved"},
    )
)
print("Reviewer sent feedback")

# --- Show full channel history ---
print(f"\n=== Channel History ({len(channel.history())} messages) ===")
for i, msg in enumerate(channel.history(), 1):
    print(
        f"  {i}. {msg.from_agent} -> {msg.to_agent} [{msg.type}] ack={msg.acknowledged}"
    )

# --- Show final content ---
print("\n=== Final Draft ===")
display(Markdown(draft.output[:600]))

# GoalAgent with ALL built-in tools (web search, code exec, etc.)
goal_agent = GoalAgent.with_builtins(
    llm=llm,
    goal=Goal(
        objective="Research the top 3 Python web frameworks and create a comparison",
        success_criteria=[
            "Covers Django, Flask, and FastAPI",
            "Includes performance characteristics",
            "Provides a recommendation based on use case",
        ],
        max_steps=4,
    ),
)

# Stream with full output at every step
print("=== GoalAgent with Web Search ===\n")
for event in goal_agent.stream():
    print_event(event)

# ReflectiveAgent with built-in tools
reflective = ReflectiveAgent.with_builtins(
    llm=llm,
    reflection_prompt="Check factual accuracy, completeness, clarity. Be strict.",
    max_reflections=2,
    quality_threshold=0.8,
)

# Stream with full output and quality scores
print("=== ReflectiveAgent with Web Search ===\n")
for event in reflective.stream(
    "Write a brief guide on deploying FastAPI to AWS Lambda"
):
    print_event(event)

# Full content production pipeline
researcher = Agent.with_builtins(
    llm=llm,
    prompt="You are a research analyst. Find key facts and data points. Be concise.",
)
analyzer = Agent(
    llm=llm,
    prompt="You are a data analyst. Identify trends and insights from research. Output bullet points.",
)
writer = Agent(
    llm=llm, prompt="You are a senior content writer. Write polished, engaging content."
)
reviewer = Agent(
    llm=llm,
    prompt="You are an editor. Review for accuracy, clarity, and engagement. Output APPROVED or list issues.",
)


# Plain Python function for word count
def add_metadata(text: str) -> str:
    words = len(text.split())
    sentences = text.count(".") + text.count("!") + text.count("?")
    return f"[{words} words, {sentences} sentences]\n\n{text}"


# Build the pipeline
pipe = Pipeline(
    # Research and analyze in parallel
    parallel(
        step(
            "research",
            agent=researcher,
            prompt="Research the current state of {topic} in 2025",
        ),
        step("trends", agent=analyzer, prompt="What are the top 3 trends in {topic}?"),
    ),
    # Write using both research and trends
    step(
        "draft",
        agent=writer,
        prompt="""Write a 3-paragraph article about {topic} using:

Research: {research.output}

Trends: {trends.output}""",
    ),
    # Add metadata (pure Python, no LLM)
    step("final", fn=add_metadata),
)

import time

start = time.time()
result = pipe.run(topic="AI-powered developer tools")
elapsed = time.time() - start

print(f"Pipeline completed in {elapsed:.1f}s\n")
print("=== Steps ===")
for name, step_result in result.steps.items():
    print(f"  {name}: {step_result.output[:80]}...")
print("\n=== Final Output ===")
display(Markdown(result.output[:1200]))