File size: 6,550 Bytes
d9aa575 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """
Agent-Q3 — Dedicated Coder Service
Qwen3-Coder-53B TOTAL-RECALL (3B active MoE, ~96%+ HumanEval)
Supported by Harmonic-Hermes for code review / second-opinion
Endpoints:
POST /v1/coder — single-pass code generation (Qwen3-Coder-53B TOTAL-RECALL)
POST /v1/coder/review — Qwen3-Coder-53B writes → Harmonic-Hermes reviews → final
GET /health
"""
from contextlib import asynccontextmanager
import httpx
import structlog
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from prometheus_fastapi_instrumentator import Instrumentator
from ..config import settings
from ..memory import memory
from ..models import HERMES_SYSTEM, ChatRequest, ChatResponse
from ..plugins import plugins
from ..router import HF_MODEL_MAP, LOCAL_MODEL_MAP, Backend, router
from ..skills import skills
log = structlog.get_logger(__name__)
# Register the dedicated coder role at module load
LOCAL_MODEL_MAP["coder_dedicated"] = lambda: settings.coder_dedicated_model
HF_MODEL_MAP["coder_dedicated"] = lambda: settings.hf_coder_model
CODER_DEDICATED_SYSTEM = """You are a senior software engineer using Qwen3-Coder-53B TOTAL-RECALL.
Write correct, idiomatic, production-grade code.
- Read the requirements precisely. Ask only if blocking ambiguity exists.
- Default to TypeScript for web, Python for scripts/ML, Solidity for contracts.
- Include only the comments that explain non-obvious WHY.
- Return working code. No filler prose, no apologies, no restating the prompt.
- If the request needs project context you don't have, state the assumption you made.
"""
@asynccontextmanager
async def lifespan(app: FastAPI):
log.info("Coder service starting", model=settings.coder_dedicated_model)
await memory.connect()
skills.load()
plugins.discover()
plugins.mount_into(app)
yield
await memory.close()
log.info("Coder service shutting down")
app = FastAPI(
title="Agent-Q3 Coder Service",
description="Dedicated Qwen3-Coder-53B TOTAL-RECALL endpoint with Harmonic-Hermes review tandem",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
Instrumentator().instrument(app).expose(app)
def _build_messages(request: ChatRequest, system: str) -> list[dict]:
msgs = []
if system and not any(m.role == "system" for m in request.messages):
msgs.append({"role": "system", "content": system})
msgs.extend([m.model_dump() for m in request.messages])
return msgs
def _extract(result: dict) -> str:
msg = result.get("message") or result.get("response") or {}
if isinstance(msg, dict):
return msg.get("content", "")
return str(msg)
@app.get("/")
async def root():
return {
"service": "Agent-Q3 Coder",
"model": settings.coder_dedicated_model,
"review_model": settings.tandem_model,
}
@app.get("/health")
async def health():
ok = False
loaded = []
try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{settings.ollama_base_url}/api/tags")
if r.status_code == 200:
ok = True
loaded = [m["name"] for m in r.json().get("models", [])]
except Exception:
pass
return {
"status": "ok" if ok else "degraded",
"ollama": ok,
"models_loaded": loaded,
"primary_coder": settings.coder_dedicated_model,
}
@app.post("/v1/coder", response_model=ChatResponse)
async def coder(request: ChatRequest):
"""Single-pass Qwen3-Coder-53B TOTAL-RECALL code generation."""
messages = _build_messages(request, CODER_DEDICATED_SYSTEM)
force_backend = Backend(request.force_backend) if request.force_backend else None
try:
result = await router.route(
model_role="coder_dedicated",
messages=messages,
force_backend=force_backend,
temperature=request.temperature,
max_tokens=request.max_tokens,
)
except Exception as e:
raise HTTPException(503, detail=str(e))
return ChatResponse(
content=_extract(result),
model_role="coder_dedicated",
model_used=result.get("_model_used", "unknown"),
backend_used=result.get("_backend_used", "unknown"),
usage=result.get("usage"),
)
@app.post("/v1/coder/review", response_model=dict)
async def coder_review(request: ChatRequest):
"""
Two-stage code generation:
Stage 1 — Qwen3-Coder-53B TOTAL-RECALL writes the code
Stage 2 — Harmonic-Hermes reviews it (correctness, edge cases, style)
"""
# Stage 1: Qwen3-Coder-53B TOTAL-RECALL
write_msgs = _build_messages(request, CODER_DEDICATED_SYSTEM)
try:
write_result = await router.route(
model_role="coder_dedicated",
messages=write_msgs,
temperature=request.temperature,
max_tokens=request.max_tokens,
)
code_out = _extract(write_result)
except Exception as e:
raise HTTPException(503, detail=f"Qwen3-Coder-53B failed: {e}")
# Stage 2: Harmonic-Hermes reviews
review_msgs = [
{"role": "system", "content": HERMES_SYSTEM},
*[m.model_dump() for m in request.messages],
{"role": "assistant", "content": f"[Qwen3-Coder-53B draft]\n{code_out}"},
{
"role": "user",
"content": "Review the code above. Identify bugs, edge cases, security issues, "
"and idiom violations. If the code is correct, say so and explain why. "
"If it needs changes, output the corrected version.",
},
]
try:
review_result = await router.route(
model_role="tandem",
messages=review_msgs,
temperature=0.4,
max_tokens=request.max_tokens,
)
review_out = _extract(review_result)
except Exception as e:
raise HTTPException(503, detail=f"Hermes review failed: {e}")
return {
"draft": code_out,
"review": review_out,
"coder_model": write_result.get("_model_used"),
"review_model": review_result.get("_model_used"),
"coder_backend": write_result.get("_backend_used"),
"review_backend": review_result.get("_backend_used"),
}
|