DeepAgent With Skills, Memory & Verification

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 IPython.display import Markdown, display
from examples.run_multi_tool_agent import build_llm_from_env

llm = build_llm_from_env("bedrock")
print("LLM ready.")

from shipit_agent import DeepAgent
from shipit_agent.stores import InMemoryMemoryStore

memory = InMemoryMemoryStore()

deep = DeepAgent.with_builtins(
    llm=llm,
    name="deep-skills-agent",
    project_root="/tmp",
    skills=["database-architect", "code-workflow-assistant"],
    auto_use_skills=True,
    memory_store=memory,
    max_iterations=12,
)

print("DeepAgent ready.")
print(f"  Skills: {[s.id for s in deep.skills]}")
print(f"  Tools: {len(deep.tools)}")
print(f"  Memory: {type(memory).__name__}")
print("  max_iterations: 12")

result = deep.run(
    """
    We are adding tenant-based billing alerts to an existing FastAPI service.

    Project context:
    - Backend: FastAPI + PostgreSQL 15
    - Current endpoint: GET /api/billing/alerts
    - Current performance: 4.8s p95
    - Schema: alerts (1M rows), tenants (5K), invoices (2M), users (50K)
    - Main query joins all four tables with no covering indexes
    - Constraint: no breaking API changes this sprint
    - Goal: reduce latency below 500ms

    Please:
    1. Diagnose the likely bottlenecks
    2. Create a step-by-step implementation plan
    3. Suggest specific index and query improvements
    4. Outline a safe rollout strategy
    """
)

print(f"Used skills: {result.metadata.get('used_skills', [])}")
print(f"Skill tools: {result.metadata.get('used_skill_tools', [])}")
print()
display(Markdown(result.output))

verified_agent = DeepAgent.with_builtins(
    llm=llm,
    name="verified-deep-agent",
    project_root="/tmp",
    skills=["security-engineer"],
    auto_use_skills=False,
    memory_store=InMemoryMemoryStore(),
    verify=True,  # ← quality gate on the output
    max_iterations=10,
)

result = verified_agent.run(
    """
    Review the security posture of a typical FastAPI + PostgreSQL deployment.
    Check for: SQL injection risks, auth middleware gaps, CORS misconfiguration,
    exposed debug endpoints, and missing rate limiting.
    Provide a prioritised remediation plan.
    """
)

print("Verification result:")
print(result.metadata.get("verification", "N/A"))
print()
display(Markdown(result.output))

reflective_agent = DeepAgent.with_builtins(
    llm=llm,
    name="reflective-deep-agent",
    project_root="/tmp",
    skills=["technical-writer"],
    auto_use_skills=False,
    reflect=True,  # ← self-critique loop
    reflect_threshold=0.8,  # ← minimum quality score
    reflect_max_iterations=3,  # ← max revision rounds
)

result = reflective_agent.run(
    """
    Write a concise technical design doc for adding WebSocket support
    to the billing-api service. Cover: protocol choice, connection lifecycle,
    auth strategy, message format, error handling, and scaling considerations.
    Keep it under 800 words.
    """
)

print(f"Reflection rounds: {getattr(result, 'iterations', 'N/A')}")
print(f"Final score: {getattr(result, 'quality_score', 'N/A')}")
print()
display(Markdown(getattr(result, "output", str(result))))

deep_chat_agent = DeepAgent.with_builtins(
    llm=llm,
    name="deep-chat-agent",
    project_root="/tmp",
    skills=["full-stack-developer", "database-architect"],
    auto_use_skills=True,
    memory_store=InMemoryMemoryStore(),
    max_iterations=10,
)

chat = deep_chat_agent.chat(session_id="deep-debug-session")

# Turn 1: Set the context
r1 = chat.send(
    """
    I'm working on billing-api (FastAPI + PostgreSQL 15).
    The GET /api/billing/alerts endpoint is slow — 4.8s p95.
    It joins alerts, tenants, invoices, and users.
    Help me diagnose the root cause.
    """
)
print("=== Turn 1 ===")
display(Markdown(r1.output))

# Turn 2: Follow up — agent remembers everything from turn 1.
r2 = chat.send(
    "Great analysis. Now give me the exact CREATE INDEX statements I should run."
)
print("=== Turn 2 (no re-asking) ===")
display(Markdown(r2.output))

# Turn 3: Follow up again.
r3 = chat.send("How should I deploy these index changes safely with zero downtime?")
print("=== Turn 3 (full context retained) ===")
display(Markdown(r3.output))

from shipit_agent import Agent

# Create specialised sub-agents.
researcher = Agent.with_builtins(
    llm=llm,
    name="researcher",
    skills=["web-scraper-pro"],
    auto_use_skills=False,
)
db_expert = Agent.with_builtins(
    llm=llm,
    name="db-expert",
    skills=["database-architect"],
    auto_use_skills=False,
)

# DeepAgent as supervisor with delegation.
supervisor = DeepAgent.with_builtins(
    llm=llm,
    name="supervisor",
    skills=["multi-agent-manager"],
    auto_use_skills=False,
    agents=[researcher, db_expert],  # ← sub-agents
    memory_store=InMemoryMemoryStore(),
    max_iterations=15,
)

print(f"Supervisor tools: {len(supervisor.tools)}")
print(f"Sub-agents: {list(supervisor.sub_agents.keys())}")
print(f"Skills: {[s.id for s in supervisor.skills]}")

result = supervisor.run(
    """
    Research the best practices for PostgreSQL query optimisation in 2024,
    then have the database expert apply those practices to our billing-api
    endpoint (GET /api/billing/alerts, 4.8s p95, joins alerts+tenants+invoices+users).
    Produce a unified recommendation.
    """
)

display(Markdown(getattr(result, "output", str(result))))

stream_agent = DeepAgent.with_builtins(
    llm=llm,
    name="stream-skills-agent",
    project_root="/tmp",
    skills=["devops-automation"],
    auto_use_skills=False,
    max_iterations=8,
)

print("Streaming events:\n")
final_output = ""
for event in stream_agent.stream(
    "Create a CI/CD pipeline plan for a FastAPI + PostgreSQL service deployed to AWS ECS."
):
    if event.type == "tool_called":
        print(f"  [tool] {event.message}")
    elif event.type == "tool_completed":
        print(f"  [done] {event.message}")
    elif event.type == "run_completed":
        final_output = event.payload.get("output", "")
        print("  [finished]")

print()
display(Markdown(final_output))