| 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 """ |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <meta charset="UTF-8"> |
| <title>DriveCore</title> |
| <style> |
| * { margin:0; padding:0; box-sizing:border-box; } |
| body { background:#07080f; font-family:sans-serif; height:100vh; display:flex; flex-direction:column; } |
| .bar { padding:10px 20px; background:#0d0f1a; border-bottom:1px solid #1e2235; display:flex; align-items:center; gap:10px; } |
| .logo { background:#6366f1; width:28px; height:28px; border-radius:6px; display:flex; align-items:center; justify-content:center; font-size:14px; } |
| .brand { color:#e2e8f0; font-weight:700; font-size:15px; } |
| .badge { margin-left:auto; font-size:10px; background:#131627; border:1px solid #1e2235; padding:3px 10px; border-radius:20px; color:#64748b; } |
| iframe { flex:1; border:none; width:100%; height:calc(100vh - 50px); } |
| </style> |
| </head> |
| <body> |
| <div class="bar"> |
| <div class="logo">🚗</div> |
| <span class="brand">DriveCore</span> |
| <span class="badge">⚡ AMD Instinct GPU + Qwen3 + LangChain</span> |
| </div> |
| <iframe src="https://drivecore.dr-coke.workers.dev/" allowfullscreen></iframe> |
| </body> |
| </html> |
| """ |
|
|
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
