Step 8: Wire the Graph

START
  │
  ▼
analyze_job
  │
  ├──────────────┐
  │              │
  ▼              ▼
tailor_resume    write_cover_letter   ← parallel execution
  │              │
  ├──────────────┘
  │
  ▼
review_application
  │
  ├── PASS ──────────────► save_outputs → END
  │
  └── NEEDS_REVISION ────► tailor_resume (loop back, max 1 retry)

# graph.py
# ==========================================
# LangGraph workflow — wires all agents together
# ==========================================
 
from typing import Literal
from langgraph.graph import StateGraph, START, END
from state import JobApplicationState
from agents.analyzer import analyze_job
from agents.tailor import tailor_resume
from agents.writer import write_cover_letter
from agents.reviewer import review_application
from utils import save_output
 
 
# ------------------------------------------
# Output node — saves all files to disk
# ------------------------------------------
 
def save_outputs(state: JobApplicationState) -> dict:
    """Save the final application materials to the output directory."""
    if state.get("tailored_resume"):
        path = save_output("tailored_resume.md", state["tailored_resume"])
        print(f"  Saved: {path}")
 
    if state.get("cover_letter"):
        path = save_output("cover_letter.md", state["cover_letter"])
        print(f"  Saved: {path}")
 
    if state.get("review"):
        path = save_output("review.md", state["review"])
        print(f"  Saved: {path}")
 
    return {}
 
 
# ------------------------------------------
# Routing function — decides pass or revise
# ------------------------------------------
 
def should_revise(state: JobApplicationState) -> Literal["save", "revise"]:
    """Check if the application passed review or needs revision."""
    # Count how many times we have reviewed (prevent infinite loops)
    # We track this by checking if the review mentions "NEEDS_REVISION"
    # and if we already have a tailored resume (meaning this is a re-review)
    if state.get("review_passed", False):
        return "save"
 
    # Allow one revision cycle, then force save
    if state.get("tailored_resume") and "REVISED" not in state.get("review", ""):
        return "revise"
 
    # Fallback — save what we have
    return "save"
 
 
# ------------------------------------------
# Build the graph
# ------------------------------------------
 
def build_graph() -> StateGraph:
    """Construct the complete job application pipeline."""
 
    builder = StateGraph(JobApplicationState)
 
    # --- Add all nodes ---
    builder.add_node("analyze", analyze_job)
    builder.add_node("tailor", tailor_resume)
    builder.add_node("writer", write_cover_letter)
    builder.add_node("review", review_application)
    builder.add_node("save", save_outputs)
 
    # --- Wire the edges ---
 
    # START → analyze (always the first step)
    builder.add_edge(START, "analyze")
 
    # analyze → tailor AND writer (parallel fan-out)
    builder.add_edge("analyze", "tailor")
    builder.add_edge("analyze", "writer")
 
    # tailor → review AND writer → review (parallel fan-in)
    builder.add_edge("tailor", "review")
    builder.add_edge("writer", "review")
 
    # review → conditional: save or revise
    builder.add_conditional_edges(
        "review",
        should_revise,
        {
            "save": "save",
            "revise": "tailor",  # Loop back to rewrite the resume
        },
    )
 
    # save → END
    builder.add_edge("save", END)
 
    return builder.compile()
 
 
# ------------------------------------------
# Export the compiled graph
# ------------------------------------------
 
app = build_graph()

# Fan-out: one node connects to two downstream nodes
builder.add_edge("analyze", "tailor")
builder.add_edge("analyze", "writer")
 
# Fan-in: two nodes connect to one downstream node
builder.add_edge("tailor", "review")
builder.add_edge("writer", "review")

builder.add_conditional_edges(
    "review",
    should_revise,
    {
        "save": "save",       # Passed → save and finish
        "revise": "tailor",   # Failed → rewrite the resume
    },
)

# test_graph.py
# ==========================================
# Test the complete pipeline with a sample job
# ==========================================
 
from graph import app
from utils import load_file
 
resume = load_file("sample_resume.md")
 
job_posting = """
Senior Python Developer — DataFlow Inc.
 
We are looking for a Senior Python Developer to join our platform team.
You will design and build core data pipeline services.
 
Requirements:
- 5+ years of professional Python experience
- Strong experience with FastAPI or Django
- Experience with distributed systems (Kafka, RabbitMQ)
- PostgreSQL and Redis proficiency
- Docker and Kubernetes experience
- CI/CD pipeline experience
 
We are a remote-first startup that values ownership and shipping fast.
"""
 
# Run the full pipeline
print("Running job application pipeline...")
print("=" * 50)
 
result = app.invoke({
    "job_posting": job_posting,
    "resume": resume,
})
 
print("\n" + "=" * 50)
print(f"Company: {result.get('company_name')}")
print(f"Title: {result.get('job_title')}")
print(f"Review passed: {result.get('review_passed')}")
print(f"\nCheck the output/ directory for your files.")

python test_graph.py

rm test_graph.py