import os, uuid, time
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import httpx
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
GROQ_URL = "https://api.groq.com/openai/v1/chat/completions"
MODEL = "llama-3.3-70b-versatile"
QWEN_PERSONA = (
"You are Qwen, the AutoPulse Bot — an expert AI safety analyst for autonomous vehicle fleets. "
"Always start with: 'Hi, I'm Qwen the AutoPulse Bot. I am now looking through your submissions on issues and I will diagnose.'"
)
async def call_groq(prompt: str) -> str:
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(
GROQ_URL,
headers={"Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json"},
json={"model": MODEL, "messages": [{"role": "system", "content": QWEN_PERSONA}, {"role": "user", "content": prompt}], "max_tokens": 2000}
)
data = resp.json()
return data["choices"][0]["message"]["content"]
class Query(BaseModel):
input: str
class ChatMessage(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
model: str = "qwen3"
messages: list[ChatMessage]
tools: list = []
tool_choice: dict = {}
def make_completion(content: str, model: str = "qwen3"):
return {
"id": f"chatcmpl-{uuid.uuid4().hex}",
"object": "chat.completion",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": content, "tool_calls": None},
"finish_reason": "stop"
}]
}
@app.post("/triage")
async def triage(query: Query):
result = await call_groq(f"Run full AV incident triage:\n1. Classify (vehicle_id, timestamp, severity P1/P2/P3)\n2. Identify affected subsystems\n3. Root cause analysis (ISO 26262, SAE J3016, NHTSA)\n4. Response plan\n5. Formal Markdown report\nEnd with '## Qwen's Personal Recommendation'\n\nIncident: {query.input}")
return {"output": result}
@app.post("/branch-debug")
async def branch_debug(query: Query):
result = await call_groq(f"Analyze this git diff and failure. Rank root cause suspects by file path, line range, mechanism, confidence. End with '## Qwen's Personal Recommendation'.\n\n{query.input}")
return {"output": result}
@app.post("/forensic")
async def forensic(query: Query):
result = await call_groq(f"Perform forensic analysis. Flag suspicious patterns, vulnerabilities, safety issues with severity. End with '## Qwen's Personal Recommendation'.\n\n{query.input}")
return {"output": result}
@app.post("/analyze")
async def analyze(query: Query):
result = await call_groq(query.input)
return {"output": result}
@app.post("/chat/completions")
async def chat_completions(req: ChatRequest):
user_input = next((m.content for m in reversed(req.messages) if m.role == "user"), "")
result = await call_groq(f"Run full AV incident triage pipeline:\n1. Classify incident\n2. Identify affected subsystems\n3. Root cause analysis\n4. Response plan\n5. Formal report\nEnd with '## Qwen's Personal Recommendation'\n\nInput: {user_input}")
return make_completion(result)
@app.post("/v1/chat/completions")
async def v1_chat_completions(req: ChatRequest):
return await chat_completions(req)
@app.get("/health")
async def health():
return {"status": "ok", "model": "groq-llama3.3-70b", "persona": "Qwen AutoPulse Bot"}
@app.get("/", response_class=HTMLResponse)
async def index():
return """
DriveCore
🚗
DriveCore
⚡ AMD Instinct GPU + Qwen3 + LangChain
"""
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)