ShipCrew: Multi-Agent Orchestration

!playwright install   

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
from shipit_agent.deep.ship_crew import (
    ShipCrew,
    ShipAgent,
    ShipTask,
    create_ship_crew,
)
from shipit_agent.deep.ship_crew.errors import (
    ShipCrewError,
    CyclicDependencyError,
)
from examples.run_multi_tool_agent import build_llm_from_env
from IPython.display import display, Markdown

llm = build_llm_from_env("bedrock")

# Create two specialized agents
researcher = ShipAgent(
    name="researcher",
    agent=Agent(
        llm=llm,
        prompt="You are a thorough researcher. Provide detailed, factual findings.",
    ),
    role="Senior Researcher",
    goal="Find comprehensive, accurate information",
    capabilities=["research", "analysis", "summarization"],
)

writer = ShipAgent(
    name="writer",
    agent=Agent(
        llm=llm,
        prompt="You are a skilled technical writer. Write clear, engaging content.",
    ),
    role="Technical Writer",
    goal="Transform research into polished content",
    capabilities=["writing", "editing", "formatting"],
)

reviewer = ShipAgent(
    name="reviewer",
    agent=Agent(
        llm=llm,
        prompt="You are a meticulous reviewer. Provide constructive feedback and ensure quality.",
    ),
    role="Content Reviewer",
    goal="Review content for accuracy and clarity",
    capabilities=["critical analysis", "feedback", "quality assurance"],
)

# Define tasks with dependencies
crew = ShipCrew(
    name="content-crew",
    coordinator_llm=llm,
    agents=[researcher, writer, reviewer],
    tasks=[
        ShipTask(
            name="research",
            description="Research the current state of AI agents in 2026. Cover key frameworks, trends, and adoption.",
            agent="researcher",
            output_key="findings",
        ),
        ShipTask(
            name="write",
            description="Write a concise 3-paragraph summary based on these findings: {findings}",
            agent="writer",
            depends_on=["research"],
            output_key="article",
        ),
        ShipTask(
            name="review",
            description="Review the article for accuracy and clarity. Provide feedback and suggest improvements.",
            agent="reviewer",
            depends_on=["write"],
            output_key="reviewed_article",
        ),
    ],
)

for event in crew.stream():
    display(Markdown(f"### Type '{event.type}' completed by "))
    display(Markdown(event.payload.get("output", "")))

result = crew.run()
print(f"Execution order: {result.execution_order}")
print(f"Total tasks:     {result.total_tasks}")
print(f"Failed tasks:    {result.failed_tasks}")
print(f"Time:            {result.metadata.get('elapsed_seconds', 'n/a')}s")

print("\n=== Research Findings (preview) ===")
print(result.task_results.get("findings", "")[:400])

print("\n=== Final Article ===")

display(Markdown(result.output))

analyst = ShipAgent(
    name="analyst",
    agent=Agent(
        llm=llm, prompt="You are a data analyst. Provide concise numerical insights."
    ),
    role="Data Analyst",
    goal="Extract data-driven insights",
)

reviewer = ShipAgent(
    name="reviewer",
    agent=Agent(
        llm=llm, prompt="You are a critical reviewer. Verify facts and flag issues."
    ),
    role="Content Reviewer",
    goal="Ensure accuracy and quality",
)

crew = ShipCrew(
    name="report-crew",
    coordinator_llm=llm,
    agents=[researcher, analyst, writer, reviewer],
    tasks=[
        ShipTask(
            name="research",
            description="Research AI agent market size and growth",
            agent="researcher",
            output_key="research",
        ),
        ShipTask(
            name="analyze",
            description="Analyze market data and provide key statistics",
            agent="analyst",
            output_key="analysis",
        ),
        ShipTask(
            name="draft",
            description="Write a report combining:\nResearch: {research}\nAnalysis: {analysis}",
            agent="writer",
            depends_on=["research", "analyze"],
            output_key="draft",
        ),
        ShipTask(
            name="review",
            description="Review this draft for accuracy: {draft}",
            agent="reviewer",
            depends_on=["draft"],
        ),
    ],
)

errors = crew.validate()
print("Valid!" if not errors else f"Errors: {errors}")

result = crew.run()
print(f"Execution order: {result.execution_order}")
print(f"Time: {result.metadata.get('elapsed_seconds', 'n/a')}s")
display(Markdown(result.output[:600]))

display(Markdown(result.output))

crew_parallel = ShipCrew(
    name="parallel-crew",
    coordinator_llm=llm,
    agents=[researcher, analyst, writer],
    tasks=[
        ShipTask(
            name="research",
            description="Research AI agent frameworks",
            agent="researcher",
            output_key="research",
        ),
        ShipTask(
            name="analyze",
            description="Analyze AI adoption trends",
            agent="analyst",
            output_key="analysis",
        ),
        ShipTask(
            name="draft",
            description="Combine:\n{research}\n{analysis}",
            agent="writer",
            depends_on=["research", "analyze"],
        ),
    ],
    process="parallel",
)

result = crew_parallel.run()
print(f"Execution order: {result.execution_order}")
print(f"Time: {result.metadata.get('elapsed_seconds', 'n/a')}s")
print("'research' and 'analyze' ran in parallel (same DAG layer)")
display(Markdown(result.output[:500]))

crew_ctx = ShipCrew(
    name="flexible-crew",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[
        ShipTask(
            name="research",
            description="Research {topic} for a {audience} audience",
            agent="researcher",
            output_key="findings",
        ),
        ShipTask(
            name="write",
            description="Write a {format} about {topic} based on: {findings}",
            agent="writer",
            depends_on=["research"],
        ),
    ],
)

result = crew_ctx.run(
    topic="quantum computing breakthroughs",
    audience="executive leadership",
    format="2-paragraph executive brief",
)
display(Markdown(result.output[:600]))

crew_hierarchical = ShipCrew(
    name="hierarchical-demo",
    coordinator_llm=llm,
    agents=[researcher, analyst, writer],
    tasks=[
        ShipTask(
            name="research",
            description="Research AI agent safety concerns",
            agent="researcher",
            output_key="research",
        ),
        ShipTask(
            name="analyze",
            description="Analyze the severity of each concern",
            agent="analyst",
            output_key="analysis",
        ),
        ShipTask(
            name="report",
            description="Write a safety report from {research} and {analysis}",
            agent="writer",
            depends_on=["research", "analyze"],
        ),
    ],
    process="hierarchical",
    max_rounds=8,
)

result = crew_hierarchical.run()
print("Mode: hierarchical")
print(f"Execution order: {result.execution_order}")
print(
    f"Tasks completed: {result.total_tasks - len(result.failed_tasks)}/{result.total_tasks}"
)
print(f"Time: {result.metadata.get('elapsed_seconds', 'n/a')}s")
display(Markdown(result.output[:600]))

crew_stream = ShipCrew(
    name="stream-demo",
    coordinator_llm=llm,
    agents=[researcher, writer],
    tasks=[
        ShipTask(
            name="research",
            description="Research top 3 AI trends for 2026",
            agent="researcher",
            output_key="findings",
        ),
        ShipTask(
            name="write",
            description="Summarize in 2 sentences: {findings}",
            agent="writer",
            depends_on=["research"],
        ),
        ShipTask(
            name="review",
            description="Expand the summary into a 3-paragraph article",
            agent="writer",
            depends_on=["write"],
        ),
        ShipTask(
            name="finalize",
            description="Polish the article and ensure it's well-formatted",
            agent="writer",
            depends_on=["review"],
        ),
        ShipTask(
            name="publish",
            description="Publish the article to the company blog",
            agent="writer",
            depends_on=["finalize"],
        ),
        ShipTask(
            name="bonus",
            description="Write a catchy headline for the article",
            agent="writer",
            depends_on=["write"],
        ),
    ],
)

print("=== Streaming Crew Events ===\n")
for event in crew_stream.stream():
    emoji = {
        "run_started": "🚀",
        "tool_called": "⚙️",
        "tool_completed": "✅",
        "tool_failed": "❌",
        "run_completed": "🏁",
    }.get(event.type, "📌")
    print(f"{emoji} [{event.type:16s}] {event.message[:100]}")
    if event.type == "run_completed":
        print("\n=== Final Output ===")
        display(Markdown(event.payload.get("output", "")))

# Summarize the stream event mix for easier debugging
from collections import Counter

event_types = [event.type for event in crew_stream.stream()]
counts = Counter(event_types)
print("Event counts:")
for event_type, count in sorted(counts.items()):
    print(f"  {event_type:16s} {count}")

security_agent = ShipAgent.from_registry("security-auditor", llm=llm)
code_reviewer_agent = ShipAgent.from_registry("code-reviewer", llm=llm)

print(f"Security Agent: {security_agent.role}")
print(f"Code Reviewer:  {code_reviewer_agent.role}")

# Security review crew
security_crew = ShipCrew(
    name="security-review",
    coordinator_llm=llm,
    agents=[security_agent, code_reviewer_agent],
    tasks=[
        ShipTask(
            name="audit",
            description="Audit this code for vulnerabilities:\n```python\nimport os\ndef run_cmd(user_input):\n    os.system(f'echo {user_input}')\n```",
            agent=security_agent.name,
            output_key="audit_findings",
        ),
        ShipTask(
            name="review",
            description="Review audit and suggest fixes: {audit_findings}",
            agent=code_reviewer_agent.name,
            depends_on=["audit"],
        ),
    ],
)

result = security_crew.run()
display(Markdown(result.output[:800]))

crew = create_ship_crew(
    coordinator_llm=llm,
    agents=[
        {
            "name": "r",
            "agent": Agent(llm=llm, prompt="You research topics."),
            "role": "Researcher",
        },
        {
            "name": "w",
            "agent": Agent(llm=llm, prompt="You write summaries."),
            "role": "Writer",
        },
    ],
    tasks=[
        {
            "name": "research",
            "description": "Research cloud computing trends",
            "agent": "r",
            "output_key": "data",
        },
        {
            "name": "write",
            "description": "Summarize: {data}",
            "agent": "w",
            "depends_on": ["research"],
        },
    ],
    name="dict-crew",
)

result = crew.run()
print(f"Factory crew result: {result.output[:300]}")

# Missing agent
bad_crew = ShipCrew(
    name="bad-crew",
    coordinator_llm=llm,
    agents=[researcher],
    tasks=[ShipTask(name="edit", description="Edit text", agent="editor")],
)
errors = bad_crew.validate()
print("Missing agent errors:")
for e in errors:
    print(f"  ❌ {e}")

# Cyclic dependency
try:
    ShipCrew(
        name="cycle",
        coordinator_llm=llm,
        agents=[researcher, writer],
        tasks=[
            ShipTask(
                name="a", description="A needs B", agent="researcher", depends_on=["b"]
            ),
            ShipTask(
                name="b", description="B needs A", agent="writer", depends_on=["a"]
            ),
        ],
    )
except CyclicDependencyError as e:
    print(f"\nCyclic dependency: {e}")

# Unknown dependency
try:
    ShipCrew(
        name="bad-dep",
        coordinator_llm=llm,
        agents=[researcher],
        tasks=[
            ShipTask(
                name="a",
                description="A",
                agent="researcher",
                depends_on=["nonexistent"],
            )
        ],
    )
except ShipCrewError as e:
    print(f"Unknown dep: {e}")

# Task with retries, timeout, extra context, and an output schema
detailed_task = ShipTask(
    name="deep-research",
    description="Research {topic} with focus on {focus_area}",
    agent="researcher",
    depends_on=[],
    output_key="deep_findings",
    output_schema={"type": "object", "required": ["summary", "risks"]},
    max_retries=3,  # retry up to 3 times on failure
    timeout_seconds=120,  # 2 minute timeout
    context={"priority": "high", "depth": "comprehensive"},
)

print(f"Task: {detailed_task.name}")
print(f"  agent:       {detailed_task.agent}")
print(f"  output_key:  {detailed_task.output_key}")
print(f"  schema:      {detailed_task.output_schema}")
print(f"  max_retries: {detailed_task.max_retries}")
print(f"  timeout:     {detailed_task.timeout_seconds}s")
print(f"  context:     {detailed_task.context}")
print(f"  depends_on:  {detailed_task.depends_on}")

# Template resolution demo
resolved = detailed_task.resolve_description(
    {
        "topic": "neural architecture search",
        "focus_area": "efficiency gains",
    }
)
print(f"\nResolved description: {resolved}")

# Safe resolution — missing vars stay as {var}
partial = detailed_task.resolve_description({"topic": "NAS"})
print(f"Partial resolution:  {partial}")

# Serialization
import json

print(f"\nSerialized:\n{json.dumps(detailed_task.to_dict(), indent=2)}")

# Round-trip
restored = ShipTask.from_dict(detailed_task.to_dict())
print(f"\nRound-trip: name={restored.name}, retries={restored.max_retries}")

from shipit_agent.costs import CostTracker, Budget

# Create a cost tracker
tracker = CostTracker(budget=Budget(max_dollars=3.00))

# Build agents with cost-tracking hooks
tracked_researcher = ShipAgent(
    name="researcher",
    agent=Agent(llm=llm, prompt="You research topics.", hooks=tracker.as_hooks()),
    role="Researcher",
)
tracked_writer = ShipAgent(
    name="writer",
    agent=Agent(llm=llm, prompt="You write summaries.", hooks=tracker.as_hooks()),
    role="Writer",
)

cost_crew = ShipCrew(
    name="cost-tracked-crew",
    coordinator_llm=llm,
    agents=[tracked_researcher, tracked_writer],
    tasks=[
        ShipTask(
            name="research",
            description="Research microservices vs monolith",
            agent="researcher",
            output_key="findings",
        ),
        ShipTask(
            name="write",
            description="Write a comparison: {findings}",
            agent="writer",
            depends_on=["research"],
        ),
    ],
)

result = cost_crew.run()

print("=== Crew + Cost Tracking ===")
print(f"Total cost:   ${tracker.total_cost:.4f}")
print(f"Total calls:  {len(tracker.breakdown())}")
print(f"Total tokens: {tracker.total_tokens}")
print("\nPer-call:")
for c in tracker.breakdown():
    print(f"  #{c['call_number']}: {c['model'][:30]:30s} ${c['cost_usd']:.4f}")
display(Markdown(result.output[:400]))