An asynchronous AI coding system that researches, plans, generates, executes, tests, and automatically repairs Python code using LangGraph, Celery, Redis, and LLMs.
This project is a distributed multi-agent coding engine designed to handle programming tasks asynchronously.
Instead of relying on a single LLM response, the system uses a structured workflow to:
- 🔎 Research the problem
- 🧠 Decompose it into implementation steps
- 💻 Generate Python code
- 🧪 Execute the generated code
- 🔍 Inspect runtime errors
- 🔧 Automatically repair failed code
- ✅ Re-test the corrected implementation
- 📊 Produce a final analysis/report
The core idea is simple:
Don't trust generated code blindly. Execute it, observe what happens, and use the real execution feedback to improve it.
┌──────────────────────┐
│ Client / Producer │
└──────────┬───────────┘
│
Celery .delay()
│
▼
┌──────────────────────┐
│ Redis │
│ Task Broker │
└──────────┬───────────┘
│
▼
┌──────────────────────────────┐
│ Celery Workers │
│ │
│ Worker 1 Worker 2 ... │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ LangGraph │
│ │
│ Research │
│ ↓ │
│ Planner │
│ ↓ │
│ Code Generation │
│ ↓ │
│ Test / Execute │
│ │ │
│ ├── ❌ Failure │
│ │ ↓ │
│ │ Error Feedback │
│ │ ↓ │
│ │ Code Repair │
│ │ ↓ │
│ └──── Re-test │
│ │
│ ↓ │
│ Analysis │
└──────────────┬───────────────┘
│
▼
┌──────────────────────────────┐
│ Persistence & Observability │
│ │
│ SQLite + LangSmith + Logs │
└──────────────────────────────┘
The LangGraph workflow consists of multiple stages.
The research stage gathers useful information before implementation.
It can use semantic memory and external research when additional information is required.
Programming Task
↓
Research
↓
Relevant Context
The planner breaks a complex programming problem into smaller implementation steps.
For example:
Build an LRU Cache
│
├── Define data structure
├── Implement get()
├── Implement put()
├── Handle eviction
└── Validate behavior
This gives the Code Agent a structured plan before implementation.
The Code Agent generates the Python implementation based on the research and plan.
The generated code is then passed to the Test Agent.
The Test Agent doesn't simply ask an LLM whether the generated code is correct.
Instead, it actually executes the generated Python program.
The workflow uses:
tempfilesubprocess.run- Return-code validation
stdoutcapturestderrcapture- Execution timeout
Simplified flow:
Generated Python Code
↓
Temporary .py file
↓
Python subprocess
↓
┌────┴────┐
↓ ↓
Success Failure
↓ ↓
Continue Capture error
↓
Code Agent
The main feature of the system is its self-healing code generation loop.
When generated code fails during execution, the runtime error is captured and returned to the Code Agent.
Generate Code
│
▼
Execute Code
│
┌─────┴─────┐
│ │
Success Failure
│ │
▼ ▼
Continue Capture
traceback
│
▼
Code Agent
│
▼
Fix the code
│
▼
Re-test
This allows the system to respond to real execution errors such as:
- Missing imports
- Type errors
- Index errors
- Runtime exceptions
- Incorrect implementation logic
- Invalid API usage
The retry process is bounded so that the workflow does not continue indefinitely.
The system uses isolated subprocess execution to verify generated Python code.
A temporary Python file is created and executed independently.
Example:
result = subprocess.run(
["python", temp_file],
capture_output=True,
text=True,
timeout=5
)The Test Agent checks:
returncode == 0
and captures:
stdout
stderr
A 5-second timeout prevents a generated program from running indefinitely.
This is subprocess-based execution and timeout protection. It should not be considered a hardened security sandbox for arbitrary untrusted code.
The system uses Celery + Redis to process tasks asynchronously.
Instead of keeping the producer blocked while the complete AI workflow executes:
Client
│
│ submit task
▼
Redis
│
▼
Celery Worker
│
▼
LangGraph
│
▼
Final Result
Tasks are placed into the queue and processed by workers.
Example:
run_agent_task.delay(
"Write a Python function to implement binary search."
)This architecture separates:
- Task submission
- Task queuing
- Task execution
- AI workflow processing
LLM APIs can temporarily reject requests because of rate limits.
The project includes mechanisms to handle HTTP 429 responses.
Tasks can be rate-limited at the worker level:
rate_limit = "15/m"
This helps control how quickly tasks are consumed.
The safe_llm_call wrapper detects rate-limit errors and waits before retrying.
LLM Request
│
├── Success ──► Continue
│
└── 429
│
▼
Read retry delay
│
▼
Wait
│
▼
Retry
This helps prevent repeated requests from immediately triggering additional rate-limit failures.
The project includes Kubernetes manifests for the distributed worker architecture.
k8s/
├── redis-deployment.yaml
├── celery-worker-deployment.yaml
└── hpa.yaml
Runs Redis inside Kubernetes and exposes it through an internal Kubernetes Service.
Celery Worker
│
▼
redis-service:6379
│
▼
Redis Pod
Defines how Celery worker containers should run.
The configuration includes:
- Worker replicas
- Container image
- Environment variables
- Kubernetes Secrets
- CPU/memory requests
- CPU/memory limits
- Liveness probes
- Readiness probes
The HPA can dynamically adjust the number of worker Pods based on CPU and memory utilization.
Low workload
↓
2 workers
High workload
↓
3 → 4 → 5 ... workers
Maximum
↓
10 workers
The current configuration defines:
Minimum replicas: 2
Maximum replicas: 10
The repository contains Kubernetes deployment and scaling configuration. Actual deployment to a Kubernetes cluster depends on the target environment.
The system was tested by dispatching 30 asynchronous programming tasks through Celery and Redis.
| Metric | Result |
|---|---|
| Tasks dispatched | 30 |
| Successful completions | 30/30 |
| Completion rate | 100% |
| Total execution time | 823.49 seconds |
| Worker concurrency | 2 threads |
| Average throughput | ~0.036 tasks/sec |
The test demonstrates that the asynchronous task pipeline successfully processed all 30 tasks in that test run without unhandled task failures.
30/30 is a task completion result for this benchmark. It is not a claim of general AI accuracy.
The generated-code pipeline was also evaluated on five custom programming tasks.
- ISO 8601 date parsing without third-party libraries
- Custom JSON encoding for
datetimeandUUID - LRU cache implementation using
OrderedDict - Semantic version comparison
- Nested bracket matching
Tasks evaluated: 5
Tasks passed: 5
Pass rate: 100%
The solutions were executed through the automated code execution/testing workflow.
This is a custom 5-task evaluation, not a standardized benchmark such as HumanEval or MBPP.
This project combines several areas of modern AI engineering:
- LLM-powered code generation
- Agentic workflows
- Multi-agent orchestration
- Prompt-based reasoning
- Self-healing workflows
- Semantic memory
- Python
- Celery
- Redis
- Asynchronous processing
- Worker pools
- Task queues
- Retry/backoff mechanisms
- Docker
- Kubernetes
- Horizontal Pod Autoscaling
- Health probes
- Resource management
- Kubernetes Secrets
- LangSmith
- Structured logging
- Workflow tracing
- Node-level execution monitoring
- Automated code execution
- Runtime validation
- Load testing
- Custom coding benchmarks
- Pass-rate measurement
| Technology | Purpose |
|---|---|
| Python 3.11 | Core application |
| LangGraph | AI workflow orchestration |
| Celery | Distributed task execution |
| Redis | Message broker / task backend |
| FAISS | Semantic vector search |
| FastEmbed | Embedding generation |
| Groq / LLM API | LLM inference |
| LangSmith | AI workflow observability |
| SQLite | State/checkpoint persistence |
| Docker | Containerization |
| Kubernetes | Container orchestration |
| Subprocess | Generated-code execution |
multi-agent-orchestrator/
│
├── k8s/
│ ├── redis-deployment.yaml
│ ├── celery-worker-deployment.yaml
│ └── hpa.yaml
│
├── main.py
│ └── LangGraph workflow and agent nodes
│
├── celery_tasks.py
│ └── Celery application and task definitions
│
├── state.py
│ └── AgentTaskState schema
│
├── memory.py
│ └── FAISS semantic memory
│
├── run_load_test.py
│ └── Distributed 30-task load test
│
├── run_accuracy_via_celery.py
│ └── Custom coding evaluation
│
├── benchmark_accuracy.py
│ └── Benchmark harness
│
├── Dockerfile
│ └── Container definition
│
├── requirements.txt
│ └── Python dependencies
│
└── README.md
- Python 3.11+
- Redis
- Groq API key
- LangSmith API key (optional)
- Docker (optional)
- Kubernetes cluster (optional)
git clone https://github.com/<your-username>/multi-agent-orchestrator.git
cd multi-agent-orchestratorpython -m venv venv
.\venv\Scripts\activatepython -m venv venv
source venv/bin/activatepip install -r requirements.txtCreate a .env file:
GROQ_API_KEY=your_groq_api_key
REDIS_URL=redis://localhost:6379/0
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=multi-agent-orchestrator
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
LANGCHAIN_API_KEY=your_langsmith_api_keyAdd .env to .gitignore:
.env
venv/
__pycache__/
*.pyc
Never commit API keys to GitHub.
Start Redis and make sure it is available at:
localhost:6379
Then start the Celery worker:
celery -A celery_tasks worker --loglevel=info --pool=threads --concurrency=2Tasks can then be dispatched asynchronously:
run_agent_task.delay(
"Write a Python function to implement binary search."
)Build the image:
docker build -t multi-agent-orchestrator:latest .Run the worker:
docker run --env-file .env multi-agent-orchestrator:latestThe Docker image is configured to start a Celery worker by default.
Apply the Redis configuration:
kubectl apply -f k8s/redis-deployment.yamlApply the Celery worker deployment:
kubectl apply -f k8s/celery-worker-deployment.yamlApply the HPA:
kubectl apply -f k8s/hpa.yamlCheck running Pods:
kubectl get podsCheck Deployments:
kubectl get deploymentsCheck autoscaling:
kubectl get hpaView worker logs:
kubectl logs <pod-name>The project supports LangSmith tracing to inspect the AI workflow.
Tracing can help monitor:
- Agent execution
- LangGraph nodes
- LLM calls
- Workflow transitions
- Latency
- Errors
- Token usage where available
This is particularly useful when debugging multi-step agentic workflows.
The system includes several mechanisms to improve reliability:
Celery and Redis separate task submission from long-running AI execution.
Generated code is executed and checked instead of being blindly trusted.
Actual runtime errors are returned to the Code Agent for correction.
The self-healing loop has retry limits to prevent infinite execution.
HTTP 429 errors can trigger controlled waiting and retry behavior.
Kubernetes liveness and readiness probes can monitor worker health.
Kubernetes CPU and memory requests/limits provide resource boundaries for worker Pods.
Problem: LLM workflows can take significant time to complete.
Solution: Celery + Redis provide asynchronous task processing.
Problem: LLM-generated code may contain runtime or implementation errors.
Solution: Execute the code and provide real traceback information back to the Code Agent.
Problem: LLM providers can temporarily reject requests.
Solution: Task pacing and retry/backoff handling.
Problem: Multiple workers need a shared task queue.
Solution: Redis acts as the broker while Celery manages distributed workers.
Problem: Multi-step agent systems can be difficult to debug.
Solution: LangSmith tracing and structured application logging.
Problem: Worker capacity may need to increase as workload grows.
Solution: Kubernetes Deployment + Horizontal Pod Autoscaler configuration.
- Evaluate on standardized benchmarks such as HumanEval
- Add stronger container-level code isolation
- Add broader automated test coverage
- Add FastAPI API layer
- Add Prometheus/Grafana monitoring
- Add CI/CD pipeline
- Add authentication and authorization
- Deploy to a managed Kubernetes cluster
- Compare multiple LLM providers
- Expand benchmark dataset
- Compare single-pass generation against self-healing generation
╔══════════════════════════════════════════╗
║ DISTRIBUTED LOAD TEST ║
╠══════════════════════════════════════════╣
║ Tasks dispatched: 30 ║
║ Successful completions: 30 / 30 ║
║ Completion rate: 100% ║
║ Execution time: 823.49 seconds ║
║ Worker concurrency: 2 ║
╚══════════════════════════════════════════╝
╔══════════════════════════════════════════╗
║ CUSTOM CODING EVALUATION ║
╠══════════════════════════════════════════╣
║ Tasks evaluated: 5 ║
║ Tasks passed: 5 ║
║ Pass rate: 100% ║
╚══════════════════════════════════════════╝
These results represent the specific test runs performed on this project. They are not intended to represent general model accuracy, production SLA, or standardized benchmark performance.
Imad Ud Din
AI Engineer | AI Agents | Generative AI | RAG | LangGraph | Python
Interested in building practical AI systems, agentic workflows, LLM applications, and AI-powered automation.
If you find this project interesting, consider giving the repository a ⭐.