CODEFORGE Docs
Full reference for the CODEFORGE multi-agent code generation system. Built on LangGraph, powered by Groq Llama 3.3 70B, with Tavily web search, LangSmith observability, and a Flowise visual canvas.
All tests pass on Python 3.10+. Run python3 tests/test_suite.py to verify your setup.
How it works in one sentence
You give CODEFORGE a coding task in plain English. Six specialized agents pass it through a state machine: Orchestrator → Planner → Coder (with web search) → Reviewer (loops back if quality < 7) → Executor → Debugger (loops back if runtime fails). You get working, reviewed, executed code.
Quick Start
-
1Clone and installPython 3.10+ required.
git clone https://github.com/vignesh2027/CODEFORGE.git cd CODEFORGE pip install -r requirements.txt -
2Configure keysCopy the example and fill in Groq + Tavily keys.
cp .env.example .env # edit .env and set: GROQ_API_KEY=gsk_... TAVILY_API_KEY=tvly-... -
3Run
python3 main.py -
4Verify with tests
python3 tests/test_suite.py # Expected: 20/20 passing · 100%
Configuration
All configuration is via a .env file in the project root. Copy .env.example to get started. The file is gitignored and never committed.
All Environment Variables
| Variable | Required | Default | Description |
|---|---|---|---|
GROQ_API_KEY |
Required | — | Groq API key. Get free at console.groq.com |
TAVILY_API_KEY |
Required | — | Tavily search key. Get free at app.tavily.com |
MODEL_NAME |
Optional | llama-3.3-70b-versatile |
Groq model to use. All agents share this model. |
TEMPERATURE |
Optional | 0.1 |
LLM temperature. Lower = more deterministic code. |
MAX_CODE_ITERATIONS |
Optional | 3 |
Max rounds Reviewer can reject before code is forced through. |
MAX_DEBUG_ATTEMPTS |
Optional | 3 |
Max Debugger attempts before giving up. |
LANGSMITH_API_KEY |
Optional | — | LangSmith personal token (lsv2_pt_*). Do NOT use a service key. |
LANGSMITH_TRACING |
Optional | false |
Set true to enable LangSmith tracing. |
LANGSMITH_PROJECT |
Optional | CODEFORGE |
Project name in LangSmith dashboard. |
FLOWISE_API_KEY |
Optional | — | Flowise API key. Only needed for flowise/create_flow.py. |
Use a personal token (lsv2_pt_*), not a service key (lsv2_sk_*). Service keys may have null workspace IDs and will return 403 on run creation.
Agents
Six agents run in sequence. Each reads the full AgentState and writes only the fields it owns. Feedback loops are handled by LangGraph conditional edges.
1. Orchestrator
Reads the raw user task and normalizes it. Calls Groq with a strict JSON-response prompt. Detects the programming language (defaults to Python if ambiguous) and produces a clean task summary for all downstream agents.
Output format
// JSON response from Groq
{
"task_summary": "Implement a BST with insert, search, delete",
"language": "python"
}
2. Planner
Turns the task summary into a numbered step-by-step implementation blueprint. Identifies required functions, classes, libraries, edge cases, and error handling requirements.
3. Coder
Writes complete, production-quality code following the plan. Uses llm.bind_tools([search_tool]) to optionally call Tavily web search (up to 3 times per iteration) when it needs current documentation. Context-aware: reads review_feedback and debug_notes to fix prior attempts.
Tools available
TavilySearch — real-time web search. Triggered automatically when the coder needs to look up an API, library method, or implementation pattern. Max 3 calls per coding iteration.
Code extraction
Code is extracted from the LLM's response using regex to find fenced code blocks (```python ... ```). If no fence is found, the full response is used as-is.
4. Reviewer
Scores code on five criteria and decides whether to approve or send back to the Coder. The routing logic in graph/workflow.py reads review_approved and either advances to Executor or loops back to Coder.
Scoring criteria
| # | Criterion | What it checks |
|---|---|---|
| 1 | Correctness | Does it actually solve the task? |
| 2 | Completeness | Is it runnable right now without modifications? |
| 3 | Code Quality | Is it clean and idiomatic for the language? |
| 4 | Error Handling | Are edge cases and invalid inputs handled? |
| 5 | Security | No obvious vulnerabilities (injection, unsafe eval, etc.)? |
Approval rule: score ≥ 7 AND no critical issues → approved: true. Below threshold → returns to Coder with specific feedback. After MAX_CODE_ITERATIONS rounds, forces approval regardless of score.
JSON response format
{
"approved": true,
"score": 9,
"issues": [],
"summary": "Clean BST implementation, all edge cases handled"
}
5. Executor
Runs the approved code in an isolated subprocess. Writes code to a temp file, executes with sys.executable, and captures stdout/stderr.
Language handling
| Language | Action |
|---|---|
| Python | Write to /tmp/*.py, run via subprocess.run([sys.executable, tmp]) |
| JavaScript / TypeScript | Marks execution_success=True with note — no runtime available |
| Java / Go / Rust / other | Same — marks success, skips execution |
The 30-second timeout is a hard kill via subprocess.run(..., timeout=30). Timeouts are caught and returned as an error to the Debugger.
6. Debugger
Receives the failed code and full error traceback. Diagnoses the root cause (not just the symptom) and returns a complete fixed version. Increments debug_attempts — when it reaches MAX_DEBUG_ATTEMPTS, the workflow ends.
Response format
DIAGNOSIS: IndentationError on line 14 — mixed tabs and spaces
```python
# complete fixed code here
```
Shared State
All agents communicate through a single AgentState TypedDict defined in state/schema.py. LangGraph passes this state through the graph — each agent receives the full object and returns only the fields it modifies.
class AgentState(TypedDict):
task: str # raw task → summarized by Orchestrator
plan: str # implementation plan from Planner
language: str # detected language (e.g. "python")
code: str # current code from Coder/Debugger
code_iteration: int # how many times Coder has run
review_feedback: str # Reviewer's rejection reason
review_approved: bool # True = proceed to Executor
execution_output: str # stdout/stderr from subprocess
execution_success: bool # True = code ran without error
debug_attempts: int # how many Debugger cycles ran
debug_notes: str # Debugger's diagnosis
current_agent: str # name of the currently active agent
agent_log: List[str] # chronological activity log
final_code: str # confirmed working code (set by Executor)
final_explanation: str # human-readable explanation
messages: Annotated[List, add_messages] # LangChain message history
Routing Logic
Two conditional edges drive the feedback loops in graph/workflow.py:
After Reviewer
def _route_after_review(state: AgentState) -> str:
if state["review_approved"] or state["code_iteration"] >= MAX_CODE_ITERATIONS:
return "executor" # approved, or max loops hit
return "coder" # send back for rework
After Executor
def _route_after_execution(state: AgentState) -> str:
if state["execution_success"] or state["debug_attempts"] >= MAX_DEBUG_ATTEMPTS:
return END # success, or max debug attempts
return "debugger" # execution failed — debug it
Tools
TavilySearch — tools/search.py
from langchain_tavily import TavilySearch
def get_search_tool():
return TavilySearch(max_results=3)
The Coder agent binds this tool with llm.bind_tools([search_tool]). The LLM decides autonomously whether to call it.
execute_python_code — tools/repl.py
def execute_python_code(code: str, timeout: int = 30) -> tuple[bool, str]:
# writes to tempfile, runs subprocess, returns (success, output)
Returns (True, stdout) on success, (False, error_message) on failure or timeout. The temp file is always cleaned up.
LangSmith Integration
LangSmith traces every LLM call in the pipeline automatically when enabled. No code changes required — tracing is controlled entirely by environment variables.
Enable tracing
# In .env
LANGSMITH_API_KEY=lsv2_pt_... # personal token, not service key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=CODEFORGE
What gets traced
- Every agent's LLM call (prompt + response)
- Token counts (input / output / total)
- Latency per call
- Tool calls (Tavily search queries + results)
- Run metadata (agent name, iteration count)
Open smith.langchain.com → Projects → CODEFORGE. Each full pipeline run appears as a top-level trace with nested agent spans.
Flowise Integration
CODEFORGE ships with a script that automatically creates and deploys a visual agent canvas in Flowise Cloud.
Deploy
# Set FLOWISE_API_KEY in .env, then:
python3 flowise/create_flow.py
The script creates a CHATFLOW with three nodes — ChatGroq → BufferMemory → CODEFORGE Agent — with real edge connections and the Groq API key pre-embedded. It prints the canvas URL when done.
Canvas structure
| Node | Type | Role |
|---|---|---|
| ChatGroq | Chat Model | Llama 3.3 70B via Groq |
| BufferMemory | Memory | Conversation history |
| CODEFORGE Agent | Tool Agent | 5-step code generation with system prompt |
Flowise Cloud returns 403 when creating credentials via API. The script embeds the Groq key directly in the node's inputs dict as a workaround. This is fine for personal use — for production, add the credential via the Flowise UI.
Test Suite
python3 tests/test_suite.py
| Category | Tests | What they cover |
|---|---|---|
| Configuration | 1 | All env vars loaded, keys present and non-empty |
| API Connectivity | 5 | Groq responds with correct model, token counting works, Tavily returns results, LangSmith project exists, runs can be created/closed |
| Core Tools | 3 | Python code executes correctly, runtime errors are caught, 30s timeout is enforced |
| Individual Agents | 6 | Orchestrator detects Python + JavaScript, Planner produces plan >100 chars, Coder generates code, Reviewer approves good code and rejects incomplete code |
| Full Pipeline | 5 | Factorial (simple), stack (data structure), bubble sort (algorithm + execution output), JS deep clone (non-Python), doubly linked list (triggers review loop) |
| Total | 20 | 100% passing |
Adding tests
def test_my_feature():
result = my_function()
assert result == expected, f"Got {result}"
return "Descriptive success message"
# Register in main():
suite.run("Category", "My feature", test_my_feature)
File Structure
CODEFORGE/
├── main.py ← entry point, run this
├── requirements.txt ← pip dependencies
├── .env.example ← copy to .env, fill in keys
├── LICENSE ← MIT
├── CONTRIBUTING.md
│
├── state/
│ └── schema.py ← AgentState TypedDict
│
├── config/
│ ├── settings.py ← env loader, LangSmith setup
│ └── llm.py ← ChatGroq(model, max_retries=3)
│
├── agents/
│ ├── orchestrator.py ← Agent 1: language detection
│ ├── planner.py ← Agent 2: implementation plan
│ ├── coder.py ← Agent 3: code generation + Tavily
│ ├── reviewer.py ← Agent 4: quality scoring
│ ├── executor.py ← Agent 5: subprocess runner
│ └── debugger.py ← Agent 6: error diagnosis
│
├── tools/
│ ├── search.py ← TavilySearch wrapper
│ └── repl.py ← execute_python_code()
│
├── graph/
│ └── workflow.py ← StateGraph, routing functions
│
├── flowise/
│ └── create_flow.py ← Flowise CHATFLOW builder
│
├── tests/
│ └── test_suite.py ← 20 tests, Rich report
│
└── docs/ ← GitHub Pages (auto-deploys)
├── index.html ← landing page
└── docs.html ← this file
FAQ
Which Groq models can I use?
Any model supported by Groq and accessible via langchain-groq. The default is llama-3.3-70b-versatile. Change it with MODEL_NAME in .env. Other tested options: llama3-8b-8192 (faster, lower quality), mixtral-8x7b-32768.
Why subprocess for code execution instead of exec() or PythonREPL?
Subprocess runs in a completely separate Python process, so a crash, infinite loop, or malicious code cannot affect the main CODEFORGE process. exec() shares the same process memory and namespace. PythonREPLTool from LangChain is a thin wrapper around exec().
Does it work with non-Python languages?
Yes — Orchestrator detects the language and Coder writes idiomatic code for it. Execution is Python-only (subprocess). JavaScript, TypeScript, Go, Rust, Java, C++, and others get full generation + review, then execution_success is set to True with a note that execution was skipped.
What happens if Groq rate-limits me?
The ChatGroq instance is configured with max_retries=3 and automatic exponential backoff. Heavy consecutive pipeline runs (5+ in a row) can exhaust the free 12,000 TPM limit. Wait 60 seconds and retry.
Can I use a different LLM provider?
Yes. Change config/llm.py to return any BaseChatModel-compatible object (ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI, etc.). All agents use the model via the same interface.