# LangGraph Multi-Agent Swarm Production Repository ==================================================================== MODULE 1: REPOSITORY MANIFEST, DOCKER & ENVIRONMENT ==================================================================== - Runtime Dependencies (requirements.txt): ```text langgraph>=0.2.28 langgraph-checkpoint-postgres>=2.0.2 psycopg[binary,pool]>=3.2.1 langchain-openai>=0.1.25 langchain-core>=0.2.35 pydantic>=2.8.2 pytest>=8.3.2 pytest-asyncio>=0.23.8 FROM python:3.11-slim AS base ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PIP_NO_CACHE_DIR=1 WORKDIR /app RUN useradd -m -u 1000 appuser && \ apt-get update && apt-get install -y --no-install-recommends curl && \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install -r requirements.txt COPY . . RUN chown -R appuser:appuser /app USER appuser CMD ["python", "agent_swarm.py"] version: '3.8' services: postgres: image: postgres:16-alpine container_name: swarm-postgres environment: POSTGRES_DB: swarm_db POSTGRES_USER: swarm_user POSTGRES_PASSWORD: swarm_password ports: - "5432:5432" volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U swarm_user -d swarm_db"] interval: 5s timeout: 5s retries: 5 swarm-agent: build: . container_name: swarm-worker depends_on: postgres: condition: service_healthy env_file: - .env environment: - DATABASE_URL=postgresql://swarm_user:swarm_password@postgres:5432/swarm_db volumes: pgdata: OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx DATABASE_URL=postgresql://swarm_user:swarm_password@localhost:5432/swarm_db LANGCHAIN_TRACING_V2=true LANGCHAIN_ENDPOINT=[https://api.smith.langchain.com](https://api.smith.langchain.com) LANGCHAIN_API_KEY=lsv2_pt_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx LANGCHAIN_PROJECT=code-remediation-swarm MAX_RECURSION_LIMIT=3 import os import sys from typing import TypedDict, Annotated, List, Literal from pydantic import BaseModel, Field from psycopg_pool import ConnectionPool from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage from langgraph.graph.message import add_messages from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver from langgraph.types import interrupt, Command from langchain_openai import ChatOpenAI # 1. State Definition class SwarmState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] code_artifact: str security_feedback: str audit_passed: bool iteration_count: int max_iterations: int deploy_confirmed: bool # 2. LLM Engine llm = ChatOpenAI(model="gpt-4o", temperature=0.1) # 3. Agent Worker Nodes def coder_node(state: SwarmState) -> dict: current_iteration = state.get("iteration_count", 0) + 1 feedback = state.get("security_feedback", "") user_request = state["messages"][0].content prompt = [ SystemMessage(content="You are a Principal Software Engineer. Write secure, production-grade Python code."), HumanMessage(content=f"Requirement: {user_request}\nAudit Feedback: {feedback}") ] response = llm.invoke(prompt) return { "code_artifact": response.content, "iteration_count": current_iteration, "messages": [AIMessage(content=f"[Coder]: Built revision {current_iteration}")] } def security_auditor_node(state: SwarmState) -> dict: code = state.get("code_artifact", "") prompt = [ SystemMessage(content="You are an Application Security Engineer. Audit the code for SQL injection, buffer overruns, and OWASP Top 10 vulnerabilities. Respond strictly with 'STATUS: PASS' if secure, or 'STATUS: FAIL: <reasons>' if insecure."), HumanMessage(content=f"Code:\n{code}") ] response = llm.invoke(prompt) audit_result = response.content passed = "STATUS: PASS" in audit_result return { "security_feedback": audit_result, "audit_passed": passed, "messages": [AIMessage(content=f"[Auditor]: {audit_result}")] } def human_approval_node(state: SwarmState) -> dict: code_to_review = state.get("code_artifact", "") decision = interrupt({ "action": "production_deployment_authorization", "code_preview": code_to_review, "message": "Security audit passed. Authorize deployment to production?" }) if isinstance(decision, dict) and decision.get("approved") is True: return { "deploy_confirmed": True, "messages": [AIMessage(content="[Human Approver]: Production deployment authorized.")] } else: return { "deploy_confirmed": False, "messages": [AIMessage(content="[Human Approver]: Production deployment rejected.")] } def deploy_node(state: SwarmState) -> dict: if not state.get("deploy_confirmed", False): return {"messages": [AIMessage(content="[Deployer]: Deployment aborted by policy.")]} return {"messages": [AIMessage(content="[Deployer]: Code successfully deployed to production.")]} # 4. Routing Logic def audit_router(state: SwarmState) -> Literal["coder_node", "human_approval_node", "__end__"]: if state["audit_passed"]: return "human_approval_node" if state["iteration_count"] >= state["max_iterations"]: return "__end__" return "coder_node" def deployment_router(state: SwarmState) -> Literal["deploy_node", "__end__"]: if state.get("deploy_confirmed", False): return "deploy_node" return "__end__" # 5. Graph Assembly def build_swarm(pool: ConnectionPool): workflow = StateGraph(SwarmState) workflow.add_node("coder_node", coder_node) workflow.add_node("security_auditor_node", security_auditor_node) workflow.add_node("human_approval_node", human_approval_node) workflow.add_node("deploy_node", deploy_node) workflow.add_edge(START, "coder_node") workflow.add_edge("coder_node", "security_auditor_node") workflow.add_conditional_edges("security_auditor_node", audit_router, { "coder_node": "coder_node", "human_approval_node": "human_approval_node", "__end__": END }) workflow.add_conditional_edges("human_approval_node", deployment_router, { "deploy_node": "deploy_node", "__end__": END }) workflow.add_edge("deploy_node", END) checkpointer = PostgresSaver(pool) checkpointer.setup() return workflow.compile(checkpointer=checkpointer) if __name__ == "__main__": db_url = os.environ.get("DATABASE_URL", "postgresql://swarm_user:swarm_password@localhost:5432/swarm_db") with ConnectionPool(conninfo=db_url, max_size=10, kwargs={"autocommit": True}) as pool: app = build_swarm(pool) thread_config = {"configurable": {"thread_id": "prod-deployment-run-001"}} initial_state = { "messages": [HumanMessage(content="Create a FastAPI route that accepts user id and executes a database lookup.")], "code_artifact": "", "security_feedback": "", "audit_passed": False, "iteration_count": 0, "max_iterations": 3, "deploy_confirmed": False } print("--- Running Swarm until HITL Breakpoint ---", file=sys.stderr) for event in app.stream(initial_state, config=thread_config): for node_name, result in event.items(): print(f"Executed: {node_name}", file=sys.stderr) state_snapshot = app.get_state(thread_config) if state_snapshot.tasks: interrupt_data = state_snapshot.tasks[0].interrupts[0].value print(f"\n[INTERRUPT TRIGGERED]: {interrupt_data['message']}") print("--- Resuming with Human Approval ---", file=sys.stderr) for event in app.stream(Command(resume={"approved": True}), config=thread_config): for node_name, result in event.items(): print(f"Executed: {node_name}", file=sys.stderr) import pytest from unittest.mock import MagicMock from langchain_core.messages import AIMessage @pytest.fixture def mock_llm_pass(monkeypatch): mock = MagicMock() mock.invoke.side_effect = [ AIMessage(content="def lookup(user_id: int): return db.query(user_id)"), AIMessage(content="STATUS: PASS - No vulnerabilities detected.") ] return mock @pytest.fixture def sample_initial_state(): return { "messages": [], "code_artifact": "", "security_feedback": "", "audit_passed": False, "iteration_count": 0, "max_iterations": 3, "deploy_confirmed": False } import pytest from agent_swarm import coder_node, security_auditor_node, audit_router def test_coder_node_increments_iteration(monkeypatch): from langchain_core.messages import HumanMessage, AIMessage fake_llm = lambda *args, **kwargs: AIMessage(content="def test(): pass") monkeypatch.setattr("agent_swarm.llm.invoke", fake_llm) state = { "messages": [HumanMessage(content="Write function")], "code_artifact": "", "security_feedback": "", "audit_passed": False, "iteration_count": 0, "max_iterations": 3, "deploy_confirmed": False } res = coder_node(state) assert res["iteration_count"] == 1 assert "def test(): pass" in res["code_artifact"] def test_audit_router_termination_on_max_iterations(): state = { "audit_passed": False, "iteration_count": 3, "max_iterations": 3 } destination = audit_router(state) assert destination == "__end__" def test_audit_router_routes_to_human_on_pass(): state = { "audit_passed": True, "iteration_count": 1, "max_iterations": 3 } destination = audit_router(state) assert destination == "human_approval_node" +-----------+ | START | +-----+-----+ | v +-------------------+ | coder_node |<--------------------+ +---------+---------+ | | | v | (audit_passed == False & +-------------------+ | iter < max_iterations) | security_auditor |---------------------+ +---------+---------+ | | (audit_passed == True) v +-------------------+ | human_approval | === (DYNAMIC INTERRUPT: approval_gate) +---------+---------+ | +--- (approved == True) ------> +-----------------+ ----> +-------+ | | deploy_node | | END | +--- (approved == False) -----> +-----------------+ ----> +-------+ # Autonomous Code Remediation & Security Audit Swarm Production multi-agent workflow engineered with LangGraph v0.2, PostgreSQL checkpointing, and dynamic human-in-the-loop approvals. ## Quickstart 1. Clone repository and set up environment: ```bash cp .env.example .env docker compose up -d postgres pip install -r requirements.txt pytest test_swarm.py -v python agent_swarm.py docker compose up -d postgres pytest test_swarm.py python agent_swarm.py