Deep Agents

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 import (
    GoalAgent,
    Goal,
    ReflectiveAgent,
    AdaptiveAgent,
    Supervisor,
    Worker,
    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")

# Define a goal with explicit success criteria
goal_agent = GoalAgent(
    llm=llm,
    goal=Goal(
        objective="Create a comprehensive comparison of Python async libraries",
        success_criteria=[
            "Covers asyncio, trio, and curio",
            "Includes pros and cons for each",
            "Provides a recommendation",
        ],
        max_steps=5,
    ),
)

result = goal_agent.run()

print(f"Goal Status:  {result.goal_status}")
print(f"Steps Taken:  {result.steps_taken}")
print(f"Criteria Met: {result.criteria_met}")

print("\n=== Step-by-Step Progress ===")
for s in result.step_outputs:
    print(f"\n  Step {s['step']}: {s['task'][:60]}...")
    print(f"  Output: {s['output'][:100]}...")

print("\n=== Final Output ===")
display(Markdown(result.output[:1000]))

reflective = ReflectiveAgent(
    llm=llm,
    reflection_prompt="Check for: accuracy, completeness, clarity, and actionability. Be critical but constructive.",
    max_reflections=3,
    quality_threshold=0.8,
)

result = reflective.run(
    "Explain the CAP theorem and its practical implications for distributed databases"
)

print(f"Final Quality: {result.final_quality}")
print(f"Iterations:    {result.iterations}")
print(f"Revisions:     {len(result.revisions)}")

print("\n=== Reflection History ===")
for i, r in enumerate(result.reflections, 1):
    print(
        f"\n  Reflection {i}: quality={r.quality_score:.2f}, needs_revision={r.revision_needed}"
    )
    print(f"  Feedback: {r.feedback[:150]}...")

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

display(Markdown(result.output))

from shipit_agent.tools.base import ToolContext

adaptive = AdaptiveAgent(llm=llm, can_create_tools=True)

# Dynamically create a Fibonacci tool
fib_tool = adaptive.create_tool(
    name="fibonacci",
    description="Calculate the Nth Fibonacci number",
    code="""
    def fibonacci(n: int) -> str:
        if n <= 1:
            return str(n)
        a, b = 0, 1
        for _ in range(2, n + 1):
            a, b = b, a + b
        return str(b)
    """,
)

# Create a temperature converter
temp_tool = adaptive.create_tool(
    name="celsius_to_fahrenheit",
    description="Convert Celsius to Fahrenheit",
    code="""
def celsius_to_fahrenheit(celsius: float) -> str:
    return f"{celsius}C = {celsius * 9/5 + 32:.1f}F"
""",
)

# Test the dynamically created tools
print("Fibonacci(10):", fib_tool.run(ToolContext(prompt="test"), n=10).text)
print("37C to F:", temp_tool.run(ToolContext(prompt="test"), celsius=37.0).text)
print(f"\nCreated tools: {[t.name for t in adaptive.created_tools]}")
print(f"Total tools available: {len(adaptive.tools)}")

# Define specialized workers
analyst = Worker(
    name="data-analyst",
    agent=Agent(
        llm=llm, prompt="You are a data analyst. Provide concise data insights."
    ),
    capabilities=["data analysis", "statistics", "trends"],
)

writer_worker = Worker(
    name="report-writer",
    agent=Agent(
        llm=llm,
        prompt="You are a business report writer. Write clear executive summaries.",
    ),
    capabilities=["writing", "reports", "summaries"],
)

# Create supervisor
supervisor = Supervisor(
    llm=llm,
    workers=[analyst, writer_worker],
    strategy="plan_and_delegate",
    max_delegations=4,
)

result = supervisor.run(
    "Analyze the trend of AI adoption in enterprises and write a brief executive summary"
)

print("=== Delegation Log ===")
for d in result.delegations:
    print(f"\n  Round {d.round}: {d.worker}")
    print(f"  Task: {d.task[:80]}...")
    print(f"  Output: {d.output[:120]}...")

print(f"\n=== Final Output (after {result.total_rounds} rounds) ===")
display(Markdown(result.output[:800]))

display(Markdown(result.output))

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

# Researcher sends findings to writer
channel.send(
    AgentMessage(
        from_agent="researcher",
        to_agent="writer",
        type="research_complete",
        data={
            "findings": [
                "AI adoption grew 35% in 2025",
                "Top use case: customer service automation",
                "Main blocker: data quality",
            ],
            "sources": 5,
            "confidence": 0.92,
        },
        requires_ack=True,
    )
)

# Writer receives the message
msg = channel.receive(agent="writer")
print(f"Writer received from {msg.from_agent}:")
print(f"  Type: {msg.type}")
print(f"  Findings: {msg.data['findings']}")
print(f"  Confidence: {msg.data['confidence']}")
print(f"  Acknowledged: {msg.acknowledged}")

# Writer acknowledges
channel.ack(msg)
print(f"  After ack: {msg.acknowledged}")

# Writer sends draft to reviewer
channel.send(
    AgentMessage(
        from_agent="writer",
        to_agent="reviewer",
        type="draft_ready",
        data={"draft": "AI adoption is accelerating..."},
    )
)

# Check pending messages
print(f"\nPending for reviewer: {channel.pending(agent='reviewer')}")
print(f"Total messages in channel: {len(channel.history())}")