What Are Deep Agents? The Architecture That Changes Everything

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import tool

llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    return f"Search results for: {query}"

tools = [search_web]
prompt = PromptTemplate.from_template(
    "Answer the following: {input}\n{agent_scratchpad}"
)

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({"input": "Analyse this bug report"})
print(result["output"])

Goal: "Review PR #847"

Plan:
├── Step 1: Fetch list of modified files from GitHub API [status: pending]
├── Step 2: Classify files by domain (auth, data layer, UI) [status: pending]
├── Step 3: Spawn security review for auth files [status: pending]
│   ├── Sub-step 3a: Review src/auth/login.py [status: pending]
│   └── Sub-step 3b: Review src/auth/tokens.py [status: pending]
├── Step 4: Spawn performance review for data layer [status: pending]
└── Step 5: Aggregate findings and post GitHub review comment [status: pending]

mkdir devpulse && cd devpulse
python -m venv venv && source venv/bin/activate
pip install langchain langchain-google-genai python-dotenv pydantic

GOOGLE_API_KEY=your_google_api_key_here
GITHUB_TOKEN=your_github_token_here  # Optional for now, we'll mock it

# 01_workspace.py
import os
import json
from pathlib import Path
from datetime import datetime
from typing import Optional

WORKSPACE_DIR = Path("./devpulse_workspace")

def init_workspace(pr_number: int) -> Path:
    """
    Create a fresh workspace directory for a PR review run.
    Returns the path to the workspace.
    """
    workspace = WORKSPACE_DIR / f"pr_{pr_number}"
    workspace.mkdir(parents=True, exist_ok=True)
    (workspace / "findings").mkdir(exist_ok=True)
    
    print(f"✅ Workspace initialized at: {workspace}")
    return workspace

def write_plan(workspace: Path, plan: dict) -> None:
    """
    Write the agent's task plan to the workspace.
    This is what allows the agent to resume if interrupted.
    """
    plan_path = workspace / "plan.json"
    plan["last_updated"] = datetime.utcnow().isoformat()
    
    with open(plan_path, "w") as f:
        json.dump(plan, f, indent=2)
    
    print(f"📝 Plan written to {plan_path}")

def read_plan(workspace: Path) -> Optional[dict]:
    """
    Read an existing plan from the workspace.
    Returns None if no plan exists (first run).
    """
    plan_path = workspace / "plan.json"
    if not plan_path.exists():
        return None
    
    with open(plan_path) as f:
        return json.load(f)

def update_task_status(workspace: Path, task_id: str, status: str, result: Optional[str] = None) -> None:
    """
    Update the status of a single task in the plan file.
    Statuses: 'pending' | 'in_progress' | 'completed' | 'failed'
    """
    plan = read_plan(workspace)
    if not plan:
        raise ValueError("No plan found. Cannot update task status.")
    
    for task in plan["tasks"]:
        if task["id"] == task_id:
            task["status"] = status
            if result:
                task["result_summary"] = result
            break
    
    write_plan(workspace, plan)
    print(f"🔄 Task '{task_id}' updated to status: {status}")

def write_finding(workspace: Path, file_path: str, findings: dict) -> None:
    """
    Write a subagent's review findings to its own file.
    Prevents different files' reviews from mixing in memory.
    """
    # Sanitize file path to create a valid filename
    safe_name = file_path.replace("/", "_").replace(".", "_") + ".json"
    finding_path = workspace / "findings" / safe_name
    
    findings["reviewed_at"] = datetime.utcnow().isoformat()
    findings["source_file"] = file_path
    
    with open(finding_path, "w") as f:
        json.dump(findings, f, indent=2)
    
    print(f"💾 Findings for '{file_path}' saved to workspace")

def read_all_findings(workspace: Path) -> list:
    """
    Read all subagent findings from the workspace findings directory.
    Used by the parent agent to aggregate results.
    """
    findings_dir = workspace / "findings"
    all_findings = []
    
    for finding_file in findings_dir.glob("*.json"):
        with open(finding_file) as f:
            all_findings.append(json.load(f))
    
    return all_findings

# 01_planner.py
import os
import json
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field
from typing import List, Literal
from pathlib import Path
from 01_workspace import init_workspace, write_plan

load_dotenv()

# ---- Data Models ----

class ReviewTask(BaseModel):
    """A single task in the PR review plan."""
    id: str = Field(description="Unique identifier, e.g. 'review_auth_py'")
    description: str = Field(description="Clear description of what this task does")
    file_path: str = Field(description="The file path this task relates to")
    review_type: Literal["security", "performance", "style", "test_coverage"] = Field(
        description="The type of review to perform"
    )
    priority: Literal["critical", "high", "medium", "low"] = Field(
        description="Priority of this task"
    )
    status: Literal["pending", "in_progress", "completed", "failed"] = Field(
        default="pending"
    )
    depends_on: List[str] = Field(
        default_factory=list,
        description="IDs of tasks that must complete before this one"
    )

class PRReviewPlan(BaseModel):
    """Complete plan for reviewing a pull request."""
    pr_number: int
    pr_title: str
    total_files: int
    tasks: List[ReviewTask]
    estimated_duration_minutes: int = Field(
        description="Rough estimate of total runtime in minutes"
    )

# ---- LLM Setup ----

llm = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash",
    temperature=0,  # Deterministic output for planning — no creativity needed here
    max_retries=3
)

# ---- Planner ----

PLANNER_SYSTEM_PROMPT = """You are the DevPulse Planning Agent.

Your ONLY job is to produce a structured review plan for a given GitHub Pull Request.

Rules:
- Break the PR into individual file review tasks
- Assign the correct review_type based on file purpose:
  * src/auth/* → security (authentication logic, token handling)
  * src/db/* or *models* → security AND performance
  * tests/* → test_coverage
  * All others → style review by default
- Mark files containing authentication, payment, or database operations as CRITICAL or HIGH priority
- Identify task dependencies (e.g., you cannot aggregate results before all reviews are complete)
- Be conservative with estimates: 2 minutes per file on average

Output ONLY valid JSON matching the PRReviewPlan schema. No explanation."""

def generate_review_plan(pr_number: int, pr_title: str, modified_files: List[str]) -> PRReviewPlan:
    """
    Call the LLM to generate a structured review plan for the given PR.
    The plan is deterministic (temperature=0) and schema-validated.
    """
    
    file_list = "\n".join(f"- {f}" for f in modified_files)
    
    messages = [
        SystemMessage(content=PLANNER_SYSTEM_PROMPT),
        HumanMessage(content=f"""
Plan a code review for this Pull Request:

PR Number: #{pr_number}
Title: {pr_title}
Modified Files ({len(modified_files)} total):
{file_list}

Generate a complete PRReviewPlan JSON object.
""")
    ]
    
    # Use structured output to get a type-safe Pydantic object back
    # This eliminates any chance of JSON parsing errors
    structured_llm = llm.with_structured_output(PRReviewPlan)
    plan = structured_llm.invoke(messages)
    
    print(f"\n✅ Planning complete. Generated {len(plan.tasks)} review tasks")
    print(f"⏱️  Estimated duration: {plan.estimated_duration_minutes} minutes")
    
    return plan

def run_planning_phase(pr_number: int, pr_title: str, modified_files: List[str]) -> dict:
    """
    Full planning phase: generates the plan and persists it to the workspace.
    Returns the workspace path for use by subsequent execution phases.
    """
    workspace = init_workspace(pr_number)
    
    print(f"\n🧠 Starting DevPulse planning phase for PR #{pr_number}...")
    print(f"   Files to review: {len(modified_files)}")
    
    # Generate the structured plan
    plan = generate_review_plan(pr_number, pr_title, modified_files)
    
    # Convert to dict and add workspace metadata before saving
    plan_dict = plan.model_dump()
    plan_dict["workspace_path"] = str(workspace)
    plan_dict["status"] = "planned"
    
    # Write to workspace — this is what enables resumability
    write_plan(workspace, plan_dict)
    
    # Print human-readable summary
    print("\n📋 Review Plan Summary:")
    for task in plan.tasks:
        priority_emoji = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"}.get(task.priority, "⚪")
        print(f"  {priority_emoji} [{task.review_type.upper()}] {task.file_path}")
    
    return plan_dict

if __name__ == "__main__":
    # Simulate a realistic PR with mixed file types
    sample_pr_files = [
        "src/auth/login.py",
        "src/auth/tokens.py",
        "src/db/user_repository.py",
        "src/api/endpoints.py",
        "src/models/user.py",
        "tests/test_auth.py",
        "tests/test_api.py",
        "README.md"
    ]
    
    plan = run_planning_phase(
        pr_number=847,
        pr_title="Refactor authentication system with JWT token rotation",
        modified_files=sample_pr_files
    )
    
    print(f"\n🎯 Plan ready. Tasks: {len(plan['tasks'])}")
    print(f"📂 Workspace: {plan['workspace_path']}")
    print("\nNext step: Execute the review plan using the DevPulse Harness (Part 2)")

✅ Workspace initialized at: devpulse_workspace/pr_847

🧠 Starting DevPulse planning phase for PR #847...
   Files to review: 8

✅ Planning complete. Generated 9 review tasks
⏱️  Estimated duration: 16 minutes

📋 Review Plan Summary:
  🔴 [SECURITY] src/auth/login.py
  🔴 [SECURITY] src/auth/tokens.py
  🔴 [SECURITY] src/db/user_repository.py
  🟠 [PERFORMANCE] src/db/user_repository.py
  🟠 [SECURITY] src/api/endpoints.py
  🟡 [STYLE] src/models/user.py
  🟢 [TEST_COVERAGE] tests/test_auth.py
  🟢 [TEST_COVERAGE] tests/test_api.py
  🟢 [STYLE] README.md

🎯 Plan ready. Tasks: 9
📂 Workspace: devpulse_workspace/pr_847

{
  "pr_number": 847,
  "pr_title": "Refactor authentication system with JWT token rotation",
  "total_files": 8,
  "tasks": [
    {
      "id": "review_auth_login_py_security",
      "description": "Check src/auth/login.py for OWASP Top 10 vulnerabilities",
      "file_path": "src/auth/login.py",
      "review_type": "security",
      "priority": "critical",
      "status": "pending",
      "depends_on": []
    },
    ...
  ],
  "estimated_duration_minutes": 16,
  "workspace_path": "devpulse_workspace/pr_847",
  "status": "planned",
  "last_updated": "2026-06-17T09:31:02.441Z"
}

# Don't do this
result = llm.invoke([HumanMessage(content=f"""
Review this entire codebase for security issues, performance problems, 
style violations, missing tests, outdated dependencies, documentation gaps,
architectural concerns, and anything else you think is important.

Here is all the code:
{entire_codebase}

Return a complete review.
""")])

# Don't do this either
messages = []
for file in pr_files:
    messages.append(HumanMessage(content=f"Now review {file}: {content}"))
    response = llm.invoke(messages)
    messages.append(response)  # Growing unboundedly

# Do this instead
plan = generate_review_plan(pr_number, pr_title, modified_files)
write_plan(workspace, plan)  # Durable state

for task in plan.tasks:
    # Each review gets a clean, isolated context
    findings = review_single_file(task)  # Part 2-3
    write_finding(workspace, task.file_path, findings)  # Durable output
    update_task_status(workspace, task.id, "completed")  # Resumability

aggregate_and_post(workspace)  # Combine only at the end