Confluence Labs scored 97.9% on ARC-AGI-2, the benchmark specifically designed to resist LLM shortcuts. Now they want to aim the same program synthesis + LLM combo at drug discovery and hardware engineering. Here's exactly how the architecture works, and whether anyone can replicate it.
TL;DR
Confluence Labs scored 97.9% on ARC-AGI-2 by framing reasoning tasks as code generation problems, 12 LLM agents write and refine Python transforms in parallel sandboxes, then majority-vote on the winner. The architecture is open-sourced; the real bet is applying it to drug discovery and hardware engineering where data is scarce and correct hypotheses are worth millions.
Build difficulty
Complete replication guide — install as a slash command or rules file
Claude's Corner# Build a Confluence Labs Clone: Program Synthesis + LLM Reasoning Engine
A step-by-step guide to build a multi-agent program synthesis system using Claude Code.
## Step 1: Set Up the Core Infrastructure
**DB Schema:**
```sql
CREATE TABLE tasks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
input_examples JSONB NOT NULL,
test_input JSONB NOT NULL,
status TEXT DEFAULT 'pending',
best_solution TEXT,
best_score FLOAT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE agent_attempts (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
task_id UUID REFERENCES tasks(id),
agent_index INT,
loop_index INT,
generated_code TEXT,
execution_output JSONB,
passed BOOLEAN,
error_message TEXT,
created_at TIMESTAMPTZ DEFAULT NOW()
);
```
## Step 2: Build the LLM Code Generator
Use the Anthropic SDK with extended thinking. The core prompt:
```python
import anthropic
def generate_transformation_code(examples, test_input, model='claude-opus-4-7'):
client = anthropic.Anthropic()
examples_str = '\n'.join([
f'Example {i+1}:\nInput: {ex["input"]}\nOutput: {ex["output"]}'
for i, ex in enumerate(examples)
])
prompt = (
'You are solving an ARC-AGI task. Study these input/output examples and write '
'Python code that transforms any input to the correct output.\n\n'
+ examples_str
+ f'\n\nTest input: {test_input}\n\n'
'Write a Python function transform(grid) that implements the pattern. '
'Then call it on the test input and print the result.'
)
response = client.messages.create(
model=model,
max_tokens=4096,
thinking={'type': 'enabled', 'budget_tokens': 2000},
system=[{'type': 'text', 'text': 'You are an expert Python programmer.', 'cache_control': {'type': 'ephemeral'}}],
messages=[{'role': 'user', 'content': prompt}]
)
return response.content[-1].text
```
Prompt caching on the system prompt cuts input token cost by ~90% on repeated calls.
## Step 3: Build the Sandboxed Code Executor
Use E2B to execute generated code safely:
```python
import e2b_code_interpreter
def execute_code_in_sandbox(code, timeout=30):
with e2b_code_interpreter.Sandbox() as sandbox:
execution = sandbox.run_code(code, timeout=timeout)
return {
'stdout': execution.logs.stdout,
'stderr': execution.logs.stderr,
'error': execution.error,
}
```
## Step 4: Implement the Refinement Loop
Each agent runs up to 10 refinement loops, feeding errors back as context:
```python
def run_agent(task_id, agent_index, examples, test_input):
previous_error = None
for loop in range(10): # MAX_REFINEMENT_LOOPS
code = generate_transformation_code(examples, test_input, error_context=previous_error)
result = execute_code_in_sandbox(code)
save_attempt(task_id, agent_index, loop, code, result)
if not result['error'] and result['stdout']:
return {'passed': True, 'code': code, 'output': result['stdout']}
previous_error = result.get('error') or result.get('stderr')
return {'passed': False, 'code': None, 'output': None}
```
## Step 5: Orchestrate 12 Parallel Agents
Run 12 agents concurrently via asyncio + ThreadPoolExecutor:
```python
import asyncio
from concurrent.futures import ThreadPoolExecutor
async def solve_task(task):
loop = asyncio.get_event_loop()
with ThreadPoolExecutor(max_workers=12) as executor:
futures = [
loop.run_in_executor(executor, run_agent,
task['id'], i,
task['input_examples'], task['test_input'])
for i in range(12)
]
results = await asyncio.gather(*futures)
passing = [r for r in results if r['passed']]
if not passing:
return None
outputs = [parse_grid(r['output']) for r in passing]
return majority_vote(outputs)
```
## Step 6: API Design
```
POST /api/tasks - Submit a reasoning task
GET /api/tasks/:id - Poll status + solution
GET /api/tasks/:id/attempts - Debug agent attempts
POST /api/tasks/:id/retry - Retry with more agents/loops
Webhook: POST /webhooks/task-complete
Body: { task_id, solution, confidence, cost_usd }
```
Rate limit: 132 concurrent sandboxes max. Queue overflow tasks in Redis.
## Step 7: Deploy and Scale
- **API**: FastAPI on Railway or Fly.io
- **Queue**: Redis + Celery, prioritize by tier (expedited/pro/free)
- **Sandboxes**: E2B managed infrastructure; fallback to AWS Lambda + Docker
- **DB**: Supabase (Postgres + RLS for multi-tenant isolation)
- **Cost strategy**: Use claude-haiku-4-5 for first 3 loops, escalate to claude-opus-4-7 only on failure
**Key insight**: Prompt caching on the few-shot examples drops input token costs by 90% when multiple agents process the same task. At $8-15/task with smart caching, this is viable for high-value scientific use cases.
Get the most important AI news daily.
