A multi-agent AI system that audits a raw data lake, designs a cleaning plan, reviews it through a committee (agents + human), generates & validates cleaning code, executes it, and produces a final quality report.
All LLM calls run through LM Studio (local). Data is plain files on disk. The pipeline is framework-agnostic and can be implemented in CrewAI, LangGraph, LangChain, or AutoGen.
- High-Level Goal
- System Architecture
- Agent Pipeline Flowchart
- Project Folder Structure
- Agent Descriptions
- Framework Comparison
- Benchmark Methodology
- Quick Start
Build a local multi-agent system that:
- Audits a
data_lake/folder of raw, messy CSV/JSON datasets. - Designs a cleaning & reorganization plan.
- Reviews the plan through a committee of agents + a human approval gate.
- Generates Python code to apply the cleaning plan.
- Validates the code via a dedicated Code Approver (loops until approved).
- Executes the approved code and saves cleaned data to
data_lake_clean/. - Evaluates the improvement and produces a final human-readable report.
The project is organized into 4 layers:
| Layer | Description |
|---|---|
| Data Layer | data_lake/ (raw inputs) → data_lake_clean/ (cleaned outputs) |
| Agent Layer | 8 specialized agents with defined roles, inputs, and outputs, framework-agnostic, callable directly or wrapped as tools |
| Shared Function Layer | shared/agent_functions.py exposes each agent's core logic as plain functions, imported identically by every orchestrator so prompts/validation never diverge |
| Orchestration Layer | Four parallel implementations of the same graph, plain Python, LangGraph, CrewAI Flow, AutoGen GroupChat, each defining ordering, parallelism, loops, and human checkpoints in that framework's native idiom |
| Interface & Config Layer | CLI (python orchestrators/run_pipeline_<framework>.py) + llm_config.yaml for LM Studio settings + benchmark.py for cross-framework comparison |
┌─────────────────────────────────────────────────────────┐
│ START │
└────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Data Explorer & Auditor Agent │
│ • Lists files & metadata │
│ • Samples rows, infers schema & types │
│ • Flags: nulls, duplicates, inconsistent formats │
│ OUTPUT → audit_report.json │
└────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Planner Agent │
│ • Proposes cleaning actions per file │
│ • Suggests schema alignment across files │
│ OUTPUT → cleaning_plan.json │
└──────────────┬──────────────────────────────────────────┘
│
┌───────┴────────┐
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Reviewer │ │ Reviewer │ (run in PARALLEL)
│ Agent 1 │ │ Agent 2 │
│ Scores & │ │ Scores & │
│ comments │ │ comments │
│ on plan │ │ on plan │
└──────┬──────┘ └──────┬──────┘
└───────┬─────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Human Approval Gate │
│ • Reviews plan + both agent reviews │
│ • Decides: APPROVE_PLAN or REQUEST_REVISION │
│ OUTPUT → approval_decision.json │
└───────────────┬────────────────────────┬────────────────┘
│ APPROVED │ REVISION REQUESTED
▼ └──────────► back to Planner
┌─────────────────────────────────────────────────────────┐
│ Coder Agent │
│ • Generates clean_data.py using pandas │
│ • Reads raw files, applies plan, writes to │
│ data_lake_clean/ │
│ OUTPUT → generated_code.py │
└────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Code Approver Agent │
│ • Checks syntax, logic, and plan alignment │
│ OUTPUT → code_review.json │
└──────────┬──────────────────────────┬───────────────────┘
│ APPROVED │ CHANGES REQUESTED
│ └────────────────────┐
│ │
│ ▼
│ ┌─────────────────────────────┐
│ │ Coder Agent (retry) │
│ │ Revises code per feedback │
│ └──────────┬──────────────────┘
│ │
│ ┌──────────┘
│ │ (loops back to Code Approver)
▼
┌─────────────────────────────────────────────────────────┐
│ Code Executor Agent │
│ • Safely runs approved generated_code.py │
│ • Monitors runtime errors │
│ OUTPUT → data_lake_clean/ + execution_log.json │
└────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Final Evaluator Agent │
│ • Re-audits cleaned files (same metrics as Explorer) │
│ • Compares before vs. after quality │
│ OUTPUT → final_report.md │
└────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ END │
└─────────────────────────────────────────────────────────┘
agentic-data-audit/
datacleaningagent/
├── audit/
│ ├── data_explorer_agent.py # 1️ Explores & audits raw data lake
│ ├── planner_agent.py # 2️ Builds the cleaning plan
│ ├── reviewer_agent.py # 3️ Two parallel reviewers score the plan
│ ├── human_in_the_loop.py # 4️ Human approve / request-revision gate
│ ├── coder_agent.py # 5️ Writes the cleaning code
│ ├── code_approver_agent.py # 6️ Validates the code (loops with coder)
│ ├── executor_agent.py # 7️ Executes approved code, writes cleaned data
│ └── final_evaluator_agent.py # 8️ Compares before/after, final report
│
├── config/
│ └── llm_config.yaml (LM Studio endpoint, model, temperature)
│
├── data_lake/
│ ├── data_lake_clean/ (cleaned versions of each dataset land here)
│ └── (raw, messy input datasets live here)
│
├── outputs/
│ ├── data_explorer_reports/ (one audit report generated per raw file)
│ ├── planner_report/ (the cleaning plan + raw model output)
│ ├── plan_reviews/ (review scores/comments from both reviewers)
│ ├── approval_gate/ (human approval decision record)
│ ├── generated_code/ (generated cleaning script + metadata)
│ ├── code_review/ (code approver's verdict + comments)
│ ├── execution/ (execution result/log after running the code)
│ └── final_evaluation/ (before/after metrics + final human-readable report)
│
├── shared/
│ ├── agent_functions.py # Thin wrappers exposing each agent's core function
│ ├── file_utils.py
│ └── metrics.py # shared quality-metric functions, used by Explorer + Evaluator
│
├── orchestrators/
│ ├── run_pipeline_plain.py # Baseline: plain Python, sequential + manual retry loops
│ ├── run_pipeline_langgraph.py # LangGraph StateGraph with conditional edges
│ ├── crewai/
│ │ └── run_pipeline_crewai.py # CrewAI Flow with @router/@listen for branching + loops
│ ├── autogen/
│ │ └── run_pipeline_autogen.py # AutoGen GroupChat with custom speaker_selection_method
│ └── benchmark.py # Prints comparison table from benchmark_results.json
│
├── .gitignore
├── requirements.txt
├── run_explorer.py
└── run_pipeline.py
Role: First contact with the raw data lake. Discovers and documents everything.
Inputs: Path to data_lake/
Responsibilities:
- List all files and their basic metadata (names, sizes, row counts).
- For each file: sample rows, infer column names and types, compute basic stats (null counts, distinct values, value distributions).
- Identify quality issues: missing data, inconsistent formats, potential duplicates, conflicting schemas.
Role: Turns the audit findings into a concrete, actionable cleaning plan.
Inputs: audit_report.json
Responsibilities:
- For each file: propose cleaning actions (drop columns, type conversions, normalization, deduplication).
- Across the whole lake: suggest schema alignment (which files can be joined or merged).
- Explain what to change, why, and how it benefits downstream analytics or AI workflows.
Role: Independent peer reviewers of the cleaning plan. Run in parallel.
Inputs: cleaning_plan.json + audit_report.json
Responsibilities:
- Independently score the plan on clarity, feasibility, and impact (1–5 scale).
- Highlight missing risks, edge cases, or alternative strategies.
Output: review_1.json, review_2.json - each containing scores and detailed comments.
Role: Final human decision point before any code is generated or executed.
Inputs: cleaning_plan.json + both review files
Responsibilities:
- Human reviews the proposed actions and the agents' concerns.
- Decides:
APPROVE_PLANorREQUEST_REVISION(with written feedback).
Output: approval_decision.json
The pipeline cannot proceed past this point without explicit human approval.
Role: Translates the approved cleaning plan into runnable Python code.
Inputs: cleaning_plan.json + approval_decision.json (only proceeds if APPROVED)
Responsibilities:
- Generate
clean_data.pyusing pandas (or similar). - Code must: read each raw file → apply plan actions → write cleaned files to
data_lake_clean/with new names.
Output: generated_code.py
The Coder Agent re-runs if the Code Approver requests changes, incorporating the provided feedback.
Role: Quality gate for the generated code. Prevents bad code from being executed.
Inputs: generated_code.py
Responsibilities:
- Check for: syntax errors, obvious logic mistakes (e.g., wrong column names vs. the plan), deviation from the approved plan.
- If problems found: produce structured feedback describing each error.
- If clean: mark as approved.
Output: code_review.json
{
"status": "changes_requested",
"comments": ["Column 'signup_dt' used but plan specifies 'signup_date'"]
}Loop: If
changes_requested, sends feedback to Coder Agent → Coder revises → Code Approver re-reviews. Repeats untilapproved.
Role: Safely runs the approved cleaning code and captures results.
Inputs: Approved generated_code.py
Responsibilities:
- Execute the script (via
subprocessor dynamic import) in a controlled environment. - Monitor for runtime errors and record full execution logs.
Output:
- Cleaned files written to
data_lake_clean/ execution_log.json(files processed, durations, errors if any)
Role: Closes the loop - measures how much the pipeline actually improved data quality.
Inputs: audit_report.json (before), data_lake_clean/ (after), execution_log.json
Responsibilities:
- Re-run the same audit metrics on cleaned files.
- Compare before vs. after: missing values, schema consistency, formatting issues, etc.
- Produce a human-friendly report explaining what improved, what remains problematic, and recommendations for future governance.
Output: final_report.md
| Framework | Native mechanism used | Fit for this pipeline |
|---|---|---|
| Plain Python | Sequential function calls + manual for loop retries |
Ground-truth baseline — simplest, most predictable, no framework overhead |
| LangGraph | StateGraph with add_conditional_edges |
Closest natural fit — the pipeline is a state machine with conditional branches |
| CrewAI | Flow with @router / @listen decorators, Crew.kickoff() for parallel reviewers |
Strong fit via Flows (not plain Crews) — router pattern maps directly onto approval/revision gates |
| AutoGen | GroupChat with a custom speaker_selection_method function overriding default "auto" routing |
Weakest natural fit — AutoGen is built for emergent conversation; deterministic ordering requires overriding its default behavior |
All four implementations call the exact same functions in shared/agent_functions.py, which in turn call the exact same prompt-building and LLM-calling code already defined in audit/*.py. Only the orchestration mechanism differs — prompts, validation logic, and the LLM are identical across all runs.
Each orchestrator appends one entry to outputs/benchmark_results.json:
{
"framework": "langgraph",
"total_duration_seconds": 142.7,
"plan_revision_count": 1,
"code_revision_count": 2,
"crashed": false
}Metrics tracked per framework:
- Total duration — end-to-end wall-clock time
- Plan revision count — how many times the Planner→Reviewers→Human loop repeated
- Code revision count — how many times the Coder→Approver loop repeated
- Crashed — whether the run threw an unrecovered exception
Run python orchestrators/benchmark.py after running all four pipelines to print a side-by-side comparison table.
# Install dependencies
pip install -r requirements.txt
pip install langgraph crewai crewai-tools pyautogen
# Run the plain-Python baseline first
python orchestrators/run_pipeline_plain.py
# Run with LangGraph
python orchestrators/run_pipeline_langgraph.py
# Run with CrewAI
python orchestrators/crewai/run_pipeline_crewai.py
# Run with AutoGen
python orchestrators/autogen/run_pipeline_autogen.py
# Compare all four runs
python orchestrators/benchmark.pyConfigure your LM Studio endpoint in config/llm_config.yaml:
endpoint: http://localhost:1234/v1
model: meta-llama-3.1-8b-instruct
temperature: 0.0This agent chain mirrors patterns used in real agentic data quality systems: discover → plan → committee → approve → implement → evaluate.