Python FastAPI PostgreSQL AI research queue
A research queue turns a question into saved evidence notes and a finished brief while the browser watches each step. Tokay runs the FastAPI page and Background Worker together, and PostgreSQL lets interrupted jobs resume from their latest checkpoint.
Last updated
You can prove the queue and recovery path without paying for a model. Three bundled documents drive the default demo, and an optional Anthropic key upgrades the same workflow to model generated notes and synthesis.
- Runtime
- Python
- Framework
- FastAPI
- Service types
- Web Service, Background Worker
- Resources
- PostgreSQL
- Required secrets
- None
What it does
The page starts with a completed heat pump brief so you can see the shape of a finished result.
Enter a research question and press Start research. The job moves through reading documents, building evidence notes, drafting the synthesis, and saving the result while the page updates live.
Check Restart the worker once during this job to watch recovery happen on purpose. The replacement process continues after the saved second step instead of repeating the job from the beginning.
Deploy it
Bring this folder to Tokay any way you like. Paste or upload it in the dashboard, push it with git, or hand it to your AI agent along with our agent instructions. Tokay figures out the rest.
Working in Claude, ChatGPT, VS Code, or another MCP client? Connect the Tokay MCP server and ask your agent to deploy this example. It signs in through your browser, so there is no token to paste.
New to Tokay? The getting started guide walks you through your first deploy.
Try it
- Submit the provided heat pump question and wait for its four steps to finish.
- Run it again with Restart the worker once during this job checked. Confirm that a later attempt resumes after step two.
- Deploy both Services again. The completed briefs and recovery evidence remain in PostgreSQL.
Add real model synthesis
Open the Background Worker's Environment page and find ANTHROPIC_API_KEY under From your code. Choose Set shared value, paste your key, and save it. Deploy the Background Worker again.
New jobs now call Anthropic once for each bundled document and once for the final synthesis. The result badge changes to Anthropic. The page keeps showing saved progress while those model calls run.
How it works
You avoid competing schema changes because the Web Service owns the Alembic migration for both processes. The Background Worker connects to those tables without trying to apply it again.
Each claimed job has a short lease that the worker renews while saving progress. If the process disappears, the lease expires and another worker continues from the checkpoint. FOR UPDATE SKIP LOCKED prevents two live workers from claiming the same job.
The queue retries a temporary failure up to three times. SSE updates the page without a refresh, while PostgreSQL remains the source of truth if that browser connection closes.
Secrets
| Name | Required | Purpose |
|---|---|---|
ANTHROPIC_API_KEY |
No | Turns the bundled document demo into model synthesis with several calls. Keep the key in encrypted Project storage. |
Grow from here
Try the PostgreSQL task tracker when you want a smaller starting point for migrations and saved CRUD data.
Source code
web/migrations/versions/0001_research_jobs.py Raw
"""Create the durable research queue.
Revision ID: 0001_research_jobs
Revises:
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = "0001_research_jobs"
down_revision: str | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.create_table(
"research_jobs",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("question", sa.Text(), nullable=False),
sa.Column("status", sa.Text(), nullable=False, server_default="queued"),
sa.Column("current_step", sa.Integer(), nullable=False, server_default="0"),
sa.Column("total_steps", sa.Integer(), nullable=False, server_default="4"),
sa.Column("checkpoint", postgresql.JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")),
sa.Column("result", postgresql.JSONB(), nullable=True),
sa.Column("locked_by", sa.Text(), nullable=True),
sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("available_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
sa.Column("last_error", sa.Text(), nullable=True),
sa.Column("restart_demo", sa.Boolean(), nullable=False, server_default=sa.text("false")),
sa.Column("restart_observed", sa.Boolean(), nullable=False, server_default=sa.text("false")),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.CheckConstraint("status IN ('queued', 'running', 'done', 'failed')", name="research_jobs_status_check"),
sa.CheckConstraint("current_step >= 0 AND current_step <= total_steps", name="research_jobs_step_check"),
)
op.create_index(
"research_jobs_claim_idx",
"research_jobs",
["status", "available_at", "created_at"],
)
def downgrade() -> None:
op.drop_index("research_jobs_claim_idx", table_name="research_jobs")
op.drop_table("research_jobs")
web/migrations/env.py Raw
import os
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
database_url = os.environ["DATABASE_URL"]
if database_url.startswith("postgresql://"):
database_url = database_url.replace("postgresql://", "postgresql+psycopg://", 1)
elif database_url.startswith("postgres://"):
database_url = database_url.replace("postgres://", "postgresql+psycopg://", 1)
config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%"))
target_metadata = None
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
web/alembic.ini Raw
[alembic]
script_location = %(here)s/migrations
prepend_sys_path = .
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
web/main.py Raw
import asyncio
import json
import os
import time
from contextlib import asynccontextmanager
from uuid import UUID, uuid4
import psycopg
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, StreamingResponse
from pydantic import BaseModel, Field
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
SEEDED_JOB_ID = UUID("00000000-0000-4000-8000-000000000017")
SEEDED_RESULT = {
"mode": "Bundled document demo",
"summary": "In this sample evidence set, heat pumps trade more retrofit planning for efficient heating and cooling. Gas furnaces can be simpler when a home already has usable gas service and ducts. Climate, electrical capacity, insulation, current distribution, local prices, and backup plans decide the better fit.",
"findings": [
{
"source": "Climate and efficiency notes",
"note": "Start with the building envelope and the heating load. Equipment labels alone do not answer how either system will perform in a specific home.",
},
{
"source": "Costs and retrofit notes",
"note": "Compare the complete project, including electrical work, ducts, gas service, cooling needs, incentives, and expected operating costs.",
},
{
"source": "Cold weather and resilience notes",
"note": "Cold climate design depends on correct sizing, low temperature capacity, defrost behavior, and a deliberate backup strategy.",
},
],
"sources": [
"climate-and-efficiency.md",
"costs-and-retrofit.md",
"cold-weather-and-resilience.md",
],
}
def connect():
return psycopg.connect(os.environ["DATABASE_URL"], row_factory=dict_row)
def seed_completed_job() -> None:
with connect() as connection:
connection.execute(
"""
INSERT INTO research_jobs
(id, question, status, current_step, total_steps, checkpoint, result, attempts)
VALUES (%s, %s, 'done', 4, 4, %s, %s, 1)
ON CONFLICT (id) DO NOTHING
""",
(
SEEDED_JOB_ID,
"Compare heat pumps and gas furnaces for an older home.",
Jsonb({"seeded": True}),
Jsonb(SEEDED_RESULT),
),
)
def wait_for_database() -> None:
delay = 1.0
for attempt in range(8):
try:
seed_completed_job()
return
except psycopg.Error:
if attempt == 7:
raise
time.sleep(delay)
delay = min(delay * 2, 8)
@asynccontextmanager
async def lifespan(_app: FastAPI):
await asyncio.to_thread(wait_for_database)
yield
app = FastAPI(title="Research queue", lifespan=lifespan)
class JobInput(BaseModel):
question: str = Field(min_length=8, max_length=500)
restart_demo: bool = False
def serialize_job(row: dict) -> dict:
return {
key: value.isoformat() if hasattr(value, "isoformat") else value
for key, value in row.items()
}
def fetch_job(job_id: UUID) -> dict | None:
with connect() as connection:
row = connection.execute(
"SELECT * FROM research_jobs WHERE id = %s",
(job_id,),
).fetchone()
return serialize_job(row) if row else None
@app.get("/health")
def health():
return {"ok": True}
@app.get("/api/jobs")
def list_jobs():
with connect() as connection:
rows = connection.execute(
"SELECT * FROM research_jobs ORDER BY created_at DESC LIMIT 30"
).fetchall()
return [serialize_job(row) for row in rows]
@app.get("/api/jobs/{job_id}")
def get_job(job_id: UUID):
job = fetch_job(job_id)
if not job:
raise HTTPException(status_code=404, detail="Job not found")
return job
@app.post("/api/jobs", status_code=202)
def create_job(job_input: JobInput):
job_id = uuid4()
with connect() as connection:
row = connection.execute(
"""
INSERT INTO research_jobs (id, question, restart_demo)
VALUES (%s, %s, %s)
RETURNING *
""",
(job_id, job_input.question.strip(), job_input.restart_demo),
).fetchone()
return serialize_job(row)
@app.get("/api/jobs/{job_id}/events")
async def job_events(job_id: UUID):
async def events():
previous = None
for _ in range(600):
job = await asyncio.to_thread(fetch_job, job_id)
if not job:
yield f"data: {json.dumps({'error': 'Job not found'})}\n\n"
return
snapshot = json.dumps(job, sort_keys=True, default=str)
if snapshot != previous:
yield f"data: {json.dumps(job, default=str)}\n\n"
previous = snapshot
if job["status"] in {"done", "failed"}:
return
await asyncio.sleep(1)
return StreamingResponse(events(), media_type="text/event-stream")
@app.get("/", response_class=HTMLResponse)
def home():
return """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Research queue</title>
<style>
:root { color-scheme: light; font-family: Inter, ui-sans-serif, system-ui, sans-serif; background: #eef2f6; color: #1d2835; }
* { box-sizing: border-box; }
body { margin: 0; }
main { width: min(1000px, calc(100% - 36px)); margin: 52px auto 90px; }
header { display: grid; grid-template-columns: 1.2fr 0.8fr; gap: 28px; padding: 38px; border-radius: 26px; background: #172a42; color: white; }
h1 { margin: 0 0 14px; font: 500 clamp(2.5rem, 7vw, 5.5rem)/0.96 Georgia, serif; }
header p { margin: 0; color: #c3d1df; line-height: 1.6; }
.steps { display: grid; align-content: center; gap: 10px; }
.steps span { padding: 10px 12px; border: 1px solid #48627d; border-radius: 10px; color: #dce7f1; }
form, article { margin-top: 24px; padding: 26px; border: 1px solid #ccd6df; border-radius: 20px; background: white; }
label { display: block; margin-bottom: 8px; font-weight: 700; }
textarea { width: 100%; min-height: 104px; resize: vertical; padding: 14px; border: 1px solid #aebbc7; border-radius: 12px; font: inherit; line-height: 1.5; }
textarea:focus { outline: 3px solid #b6d2ec; border-color: #315f8b; }
.form-row { display: flex; align-items: center; justify-content: space-between; gap: 16px; margin-top: 14px; }
.check { display: flex; align-items: center; gap: 8px; margin: 0; font-weight: 500; color: #4e6072; }
button { border: 0; border-radius: 12px; padding: 12px 18px; background: #b34e35; color: white; font: inherit; font-weight: 700; cursor: pointer; }
button:disabled { opacity: 0.55; cursor: wait; }
#jobs { display: grid; gap: 18px; }
.job-head { display: flex; align-items: start; justify-content: space-between; gap: 16px; }
h2 { margin: 0; font-size: 1.2rem; }
.status { flex: none; padding: 6px 9px; border-radius: 999px; background: #e7edf2; color: #526373; font-size: 0.78rem; font-weight: 800; text-transform: uppercase; letter-spacing: 0.05em; }
.done { background: #dceee3; color: #26603a; }
.failed { background: #f7dfdc; color: #8b3328; }
.progress { height: 8px; margin: 18px 0; border-radius: 999px; background: #e9edf1; overflow: hidden; }
.progress span { display: block; height: 100%; background: #3e779f; transition: width 300ms ease; }
.meta, .question { color: #5b6b79; line-height: 1.55; }
.result { margin-top: 18px; padding-top: 18px; border-top: 1px solid #e1e6ea; }
.result h3 { margin: 0 0 8px; }
.result li { margin: 8px 0; line-height: 1.5; }
.mode { display: inline-block; margin-bottom: 10px; color: #8a4b2f; font-weight: 700; }
@media (max-width: 700px) { main { margin-top: 18px; } header { grid-template-columns: 1fr; padding: 26px 22px; } form, article { padding: 21px; } .form-row { align-items: stretch; flex-direction: column; } button { width: 100%; } }
</style>
</head>
<body>
<main>
<header>
<div>
<h1>Research that keeps going.</h1>
<p>Submit a question, close the tab, or restart the worker. The job, lease, checkpoint, and result live in PostgreSQL.</p>
</div>
<div class="steps"><span>1 · Read bundled documents</span><span>2 · Build evidence notes</span><span>3 · Draft the synthesis</span><span>4 · Save the result</span></div>
</header>
<form id="research-form">
<label for="question">What should the worker research?</label>
<textarea id="question" minlength="8" maxlength="500">What should a homeowner compare before replacing a gas furnace with a heat pump?</textarea>
<div class="form-row">
<label class="check"><input id="restart" type="checkbox"> Restart the worker once during this job</label>
<button id="submit" type="submit">Start research</button>
</div>
</form>
<section id="jobs" aria-live="polite"></section>
</main>
<script>
const jobsElement = document.querySelector('#jobs');
const form = document.querySelector('#research-form');
const submit = document.querySelector('#submit');
const streams = new Map();
function escapeHtml(value) {
const element = document.createElement('div');
element.textContent = value == null ? '' : String(value);
return element.innerHTML;
}
function renderResult(result) {
if (!result) return '';
const findings = (result.findings || []).map(item => '<li><strong>' + escapeHtml(item.source) + '</strong> · ' + escapeHtml(item.note) + '</li>').join('');
return '<div class="result"><span class="mode">' + escapeHtml(result.mode) + '</span><h3>Brief</h3><p>' + escapeHtml(result.summary) + '</p><ul>' + findings + '</ul></div>';
}
function renderJob(job) {
let article = document.querySelector('[data-job="' + job.id + '"]');
if (!article) {
article = document.createElement('article');
article.dataset.job = job.id;
jobsElement.prepend(article);
}
const percent = Math.round((job.current_step / job.total_steps) * 100);
const recovery = job.restart_observed ? ' · worker restart observed' : '';
article.innerHTML = '<div class="job-head"><h2>' + escapeHtml(job.question) + '</h2><span class="status ' + job.status + '">' + escapeHtml(job.status) + '</span></div>' +
'<div class="progress"><span style="width:' + percent + '%"></span></div>' +
'<p class="meta">Step ' + job.current_step + ' of ' + job.total_steps + ' · attempt ' + job.attempts + recovery + '</p>' +
(job.last_error ? '<p class="meta">Last error · ' + escapeHtml(job.last_error) + '</p>' : '') + renderResult(job.result);
}
function watch(job) {
if (streams.has(job.id) || job.status === 'done' || job.status === 'failed') return;
const stream = new EventSource('/api/jobs/' + job.id + '/events');
streams.set(job.id, stream);
stream.onmessage = event => {
const current = JSON.parse(event.data);
renderJob(current);
if (current.status === 'done' || current.status === 'failed') {
stream.close();
streams.delete(job.id);
}
};
stream.onerror = () => {
stream.close();
streams.delete(job.id);
setTimeout(loadJobs, 1500);
};
}
async function loadJobs() {
const response = await fetch('/api/jobs');
const jobs = await response.json();
jobs.slice().reverse().forEach(renderJob);
jobs.forEach(watch);
}
form.addEventListener('submit', async event => {
event.preventDefault();
submit.disabled = true;
try {
const response = await fetch('/api/jobs', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
question: document.querySelector('#question').value,
restart_demo: document.querySelector('#restart').checked
})
});
if (!response.ok) throw new Error('The job could not be created.');
const job = await response.json();
renderJob(job);
watch(job);
} finally {
submit.disabled = false;
}
});
loadJobs();
</script>
</body>
</html>"""
web/requirements.txt Raw
fastapi==0.139.2
uvicorn[standard]==0.51.0
psycopg[binary]==3.3.4
alembic==1.18.5
worker/documents/climate-and-efficiency.md Raw
# Climate and efficiency notes
This sample brief begins with the heating load, not the equipment label. Insulation, air sealing, window area, floor plan, and local winter design temperature determine how much heat the home needs. A contractor should measure or calculate that load before choosing capacity.
An air source heat pump moves heat instead of creating it through combustion. Its delivered heat and efficiency change with outdoor temperature. A cold climate model should be compared at the temperatures the home will actually see, including its capacity and coefficient of performance near the local design point.
A gas furnace creates heat by burning fuel. Its rated efficiency describes conversion at the appliance, but the whole system also includes duct leakage, fan energy, and the condition of the distribution system. Existing ducts may need work for either replacement.
A heat pump can provide both heating and cooling. A furnace comparison should therefore include the separate air conditioner the home would otherwise need.
worker/documents/cold-weather-and-resilience.md Raw
# Cold weather and resilience notes
Cold weather performance is a design question. Check the heat pump's published capacity at the local winter design temperature and match that against the home's load. Oversizing and undersizing can both create comfort or operating problems.
Heat pumps periodically defrost their outdoor coils. Good controls account for that cycle. In very cold conditions, a system may use electric resistance, an existing furnace, or another backup source. The owner should understand when backup starts and what it costs to run.
Both furnaces and most heat pumps need electricity for controls and fans. A gas furnace is not automatically available during an outage. Any resilience plan should identify which loads a generator or battery can actually support.
Comfort also depends on distribution. Heat pumps often deliver air for longer periods at a lower temperature than a furnace. Duct placement, airflow, room balance, and occupant expectations deserve attention before installation.
worker/documents/costs-and-retrofit.md Raw
# Costs and retrofit notes
Compare the complete project rather than one equipment price. A heat pump proposal may include an electrical panel review, wiring, a new air handler, duct changes, condensate handling, and removal or retention of the old system. A furnace proposal may include gas service, venting, combustion air, duct repair, and a separate cooling system.
Operating cost depends on local electricity and gas prices, seasonal equipment performance, weather, thermostat choices, and the building envelope. A useful estimate shows its assumptions instead of promising one universal winner.
Incentives can change the first cost, but eligibility and value vary by place and time. Treat them as a checked input to the decision, not a number embedded permanently in an estimate.
Maintenance plans matter too. Ask who can service the selected equipment, what routine care it needs, which parts are commonly stocked, and how warranty labor is handled.
worker/main.py Raw
import os
import signal
import socket
import threading
import time
from pathlib import Path
from uuid import UUID
import psycopg
from anthropic import Anthropic
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
LEASE_SECONDS = 8
HEARTBEAT_SECONDS = 2
WORKER_ID = f"{socket.gethostname()}:{os.getpid()}"
DOCUMENT_DIRECTORY = Path(__file__).parent / "documents"
# The key is optional. Without it, the worker uses the bundled document demo.
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY", "").strip()
STOP = threading.Event()
def connect():
return psycopg.connect(os.environ["DATABASE_URL"], row_factory=dict_row)
def wait_for_database() -> None:
delay = 1.0
while not STOP.is_set():
try:
with connect() as connection:
connection.execute("SELECT 1 FROM research_jobs LIMIT 1")
return
except psycopg.Error as error:
print(f"Database not ready, retrying in {delay:.0f}s · {error}", flush=True)
STOP.wait(delay)
delay = min(delay * 2, 12)
def reclaim_expired_jobs() -> int:
with connect() as connection:
rows = connection.execute(
"""
UPDATE research_jobs
SET status = 'queued', locked_by = NULL, lease_expires_at = NULL,
available_at = now(), updated_at = now(),
last_error = 'Worker lease expired. Resuming from the last checkpoint.'
WHERE status = 'running' AND lease_expires_at < now()
RETURNING id
"""
).fetchall()
return len(rows)
def claim_job() -> dict | None:
with connect() as connection:
row = connection.execute(
"""
SELECT * FROM research_jobs
WHERE status = 'queued' AND available_at <= now()
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
"""
).fetchone()
if not row:
return None
claimed = connection.execute(
"""
UPDATE research_jobs
SET status = 'running', locked_by = %s,
lease_expires_at = now() + make_interval(secs => %s),
attempts = attempts + 1, updated_at = now()
WHERE id = %s
RETURNING *
""",
(WORKER_ID, LEASE_SECONDS, row["id"]),
).fetchone()
return claimed
class LeaseHeartbeat:
def __init__(self, job_id: UUID):
self.job_id = job_id
self.stop = threading.Event()
self.thread = threading.Thread(target=self.run, daemon=True)
def __enter__(self):
self.thread.start()
return self
def __exit__(self, _error_type, _error, _traceback):
self.stop.set()
self.thread.join(timeout=HEARTBEAT_SECONDS + 1)
def run(self):
while not self.stop.wait(HEARTBEAT_SECONDS):
try:
with connect() as connection:
connection.execute(
"""
UPDATE research_jobs
SET lease_expires_at = now() + make_interval(secs => %s), updated_at = now()
WHERE id = %s AND status = 'running' AND locked_by = %s
""",
(LEASE_SECONDS, self.job_id, WORKER_ID),
)
except psycopg.Error as error:
print(f"Heartbeat failed for {self.job_id} · {error}", flush=True)
def update_checkpoint(job_id: UUID, step: int, checkpoint: dict) -> None:
with connect() as connection:
connection.execute(
"""
UPDATE research_jobs
SET current_step = %s, checkpoint = %s, updated_at = now()
WHERE id = %s AND locked_by = %s
""",
(step, Jsonb(checkpoint), job_id, WORKER_ID),
)
def load_documents() -> list[dict]:
return [
{"source": path.name, "text": path.read_text(encoding="utf-8")}
for path in sorted(DOCUMENT_DIRECTORY.glob("*.md"))
]
def local_evidence_notes(documents: list[dict]) -> list[dict]:
notes = []
for document in documents:
paragraphs = [part.strip() for part in document["text"].split("\n\n") if part.strip() and not part.startswith("#")]
notes.append({
"source": document["source"],
"note": " ".join(paragraphs[:2]),
})
return notes
def model_evidence_notes(question: str, documents: list[dict]) -> list[dict]:
client = Anthropic(api_key=ANTHROPIC_API_KEY)
model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6")
notes = []
for document in documents:
message = client.messages.create(
model=model,
max_tokens=450,
system="Extract only evidence relevant to the question. The document is the full evidence boundary. Write a compact note and do not invent citations.",
messages=[{
"role": "user",
"content": f"Question\n{question}\n\nDocument\n{document['text']}",
}],
)
text = "".join(block.text for block in message.content if block.type == "text")
notes.append({"source": document["source"], "note": text.strip()})
return notes
def local_synthesis(question: str, notes: list[dict]) -> str:
source_names = ", ".join(note["source"].removesuffix(".md").replace("-", " ") for note in notes)
return (
f"This bundled document demo reviewed {source_names} for the question “{question}” "
"The notes agree that the decision should begin with the home, climate, existing systems, complete project cost, and a plan for the coldest conditions. "
"Use the findings below as a comparison checklist, then replace the bundled documents when you adapt this worker to your own research."
)
def model_synthesis(question: str, notes: list[dict]) -> str:
client = Anthropic(api_key=ANTHROPIC_API_KEY)
model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6")
evidence = "\n\n".join(f"Source: {note['source']}\n{note['note']}" for note in notes)
message = client.messages.create(
model=model,
max_tokens=700,
system="Write a concise research brief from only the supplied evidence. State uncertainty plainly. Do not invent sources.",
messages=[{"role": "user", "content": f"Question\n{question}\n\nEvidence notes\n{evidence}"}],
)
return "".join(block.text for block in message.content if block.type == "text").strip()
def mark_restart_observed(job_id: UUID) -> None:
with connect() as connection:
connection.execute(
"""
UPDATE research_jobs
SET restart_observed = true, updated_at = now()
WHERE id = %s AND locked_by = %s
""",
(job_id, WORKER_ID),
)
def complete_job(job_id: UUID, checkpoint: dict, result: dict) -> None:
with connect() as connection:
connection.execute(
"""
UPDATE research_jobs
SET status = 'done', current_step = total_steps, checkpoint = %s, result = %s,
locked_by = NULL, lease_expires_at = NULL, last_error = NULL, updated_at = now()
WHERE id = %s AND locked_by = %s
""",
(Jsonb(checkpoint), Jsonb(result), job_id, WORKER_ID),
)
def fail_or_retry_job(job: dict, error: Exception) -> None:
failed = job["attempts"] >= 3
status = "failed" if failed else "queued"
delay = min(job["attempts"] * 5, 20)
with connect() as connection:
connection.execute(
"""
UPDATE research_jobs
SET status = %s, locked_by = NULL, lease_expires_at = NULL,
available_at = now() + make_interval(secs => %s),
last_error = %s, updated_at = now()
WHERE id = %s AND locked_by = %s
""",
(status, delay, str(error)[:1000], job["id"], WORKER_ID),
)
def process_job(job: dict) -> None:
job_id = job["id"]
checkpoint = dict(job["checkpoint"] or {})
model_enabled = bool(ANTHROPIC_API_KEY)
print(f"Claimed {job_id} at step {job['current_step']} · attempt {job['attempts']}", flush=True)
with LeaseHeartbeat(job_id):
if job["current_step"] < 1:
documents = load_documents()
checkpoint["documents"] = documents
checkpoint["mode"] = "Anthropic" if model_enabled else "Bundled document demo"
update_checkpoint(job_id, 1, checkpoint)
job["current_step"] = 1
print(f"{job_id} · step 1 of 4 · read {len(documents)} documents", flush=True)
time.sleep(1)
if job["current_step"] < 2:
documents = checkpoint["documents"]
notes = model_evidence_notes(job["question"], documents) if model_enabled else local_evidence_notes(documents)
checkpoint["notes"] = notes
update_checkpoint(job_id, 2, checkpoint)
job["current_step"] = 2
print(f"{job_id} · step 2 of 4 · saved evidence notes", flush=True)
time.sleep(1)
if job["restart_demo"] and not job["restart_observed"]:
mark_restart_observed(job_id)
print(f"{job_id} · restart demo · exiting after checkpoint 2", flush=True)
os._exit(23)
if job["current_step"] < 3:
notes = checkpoint["notes"]
summary = model_synthesis(job["question"], notes) if model_enabled else local_synthesis(job["question"], notes)
checkpoint["summary"] = summary
update_checkpoint(job_id, 3, checkpoint)
job["current_step"] = 3
print(f"{job_id} · step 3 of 4 · saved draft", flush=True)
time.sleep(1)
result = {
"mode": checkpoint["mode"],
"summary": checkpoint["summary"],
"findings": checkpoint["notes"],
"sources": [document["source"] for document in checkpoint["documents"]],
}
checkpoint["completed"] = True
complete_job(job_id, checkpoint, result)
print(f"{job_id} · step 4 of 4 · done", flush=True)
def stop_worker(_signal_number, _frame):
STOP.set()
signal.signal(signal.SIGTERM, stop_worker)
signal.signal(signal.SIGINT, stop_worker)
def main():
wait_for_database()
print(f"Research worker ready · {WORKER_ID}", flush=True)
while not STOP.is_set():
try:
reclaimed = reclaim_expired_jobs()
if reclaimed:
print(f"Reclaimed {reclaimed} expired job lease", flush=True)
job = claim_job()
if not job:
STOP.wait(1)
continue
try:
process_job(job)
except Exception as error:
print(f"Job {job['id']} failed · {error}", flush=True)
fail_or_retry_job(job, error)
except psycopg.Error as error:
print(f"Queue loop database error · {error}", flush=True)
STOP.wait(2)
print("Research worker stopped", flush=True)
if __name__ == "__main__":
main()
worker/requirements.txt Raw
psycopg[binary]==3.3.4
anthropic==0.117.0
Built from revision 52c7b7f