LangGraph Multi-Agent System

Agentic AI workflows using LangGraph for automated data pipeline generation and code review. Multiple specialized agents collaborate to accomplish complex tasks with tool use and state management.

Architecture

The system orchestrates five specialized agents through a directed graph workflow:

                    +----------------+
                    |  Orchestrator  |
                    +-------+--------+
                            |
              +-------------+-------------+
              |             |             |
       +------+-----+ +----+----+ +------+------+
       |  Research   | |  Code   | |    Data     |
       |   Agent     | |  Gen    | |  Pipeline   |
       +-------------+ +----+----+ +-------------+
                            |
                      +-----+------+
                      |   Code     |
                      |  Reviewer  |
                      +------------+

• ResearchAgent - Searches the web, gathers information, and summarizes findings for downstream agents
• CodeGeneratorAgent - Generates code based on specifications and research context
• CodeReviewerAgent - Reviews generated code for bugs, security issues, and best practices
• DataPipelineAgent - Generates ETL pipeline code from natural language descriptions
• OrchestratorAgent - Coordinates between agents, manages workflow state and routing decisions

Features

• Stateful Workflows: TypedDict-based state management with message accumulation and checkpoint serialization
• Conditional Routing: Dynamic edge routing based on agent outputs and task requirements
• Human-in-the-Loop: Configurable checkpoints for human review before critical operations
• Code Generation + Review Cycle: Iterative subgraph that refines code through automated review
• Tool Integration: Web search, code execution sandbox, file I/O, and linting tools
• Streaming API: Real-time streaming of agent conversations via Server-Sent Events
• Error Recovery: Retry logic with exponential backoff and graceful degradation

Quick Start

# Install dependencies
pip install -r requirements.txt

# Set environment variables
export OPENAI_API_KEY="your-key"
export TAVILY_API_KEY="your-key"

# Run the API server
uvicorn src.api:app --host 0.0.0.0 --port 8000

# Submit a task
curl -X POST http://localhost:8000/task \
  -H "Content-Type: application/json" \
  -d '{"description": "Build an ETL pipeline that reads CSV files, cleans data, and loads into PostgreSQL"}'

API Endpoints

Method	Endpoint	Description
POST	/task	Submit a task for multi-agent processing
POST	/task/stream	Stream agent conversation in real-time (SSE)
GET	/task/{id}/status	Check task status and retrieve results
GET	/health	Health check endpoint

Configuration

Edit configs/config.yaml to adjust agent models, temperature settings, workflow parameters, and tool configurations. Each agent can be independently configured with different models and parameters.

Project Structure

langgraph-multi-agent/
├── src/
│   ├── agents.py          # Agent definitions and system prompts
│   ├── tools.py           # Custom tool implementations
│   ├── graph.py           # LangGraph workflow definition
│   ├── state.py           # State schema and reducers
│   └── api.py             # FastAPI application
├── configs/
│   └── config.yaml        # Agent and workflow configuration
├── requirements.txt
└── README.md

How It Works

1. A task is submitted via the API with a natural language description
2. The Orchestrator analyzes the task and determines which agents to invoke
3. Agents execute in the order defined by the graph, passing state between nodes
4. The code generation + review cycle iterates until quality thresholds are met or max iterations reached
5. Final output includes generated code, review feedback, and execution results

License

MIT License