File size: 5,463 Bytes
945aae0 | 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 | import os
import logging
from fastapi import FastAPI, Depends, HTTPException, Header
from qkernel import QKernel
from qkernel.auth import verify_token
from qkernel.memory import MemoryEngine
from qkernel.policy import PolicyEngine
from qkernel.observability import ObservabilityBrain
from qkernel.event_bus import EventBus
from qkernel.healing import SelfHealingEngine
from qkernel.executors import ChatExecutor, MemoryExecutor, ToolExecutor
from qkernel.queue import PersistentQueue
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("brain_api")
app = FastAPI(title="Q Brain API")
_queue = PersistentQueue()
def require_auth(authorization: str | None = Header(None)) -> dict:
if not authorization:
raise HTTPException(status_code=401, detail="missing_authorization")
token = authorization.removeprefix("Bearer ")
payload = verify_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="invalid_or_expired_token")
return payload
def optional_auth(authorization: str | None = Header(None)) -> dict | None:
if not authorization:
return None
token = authorization.removeprefix("Bearer ")
return verify_token(token)
def build_kernel() -> QKernel:
kernel = QKernel()
from qkernel.policy import PolicyRule
kernel.policy.add_rule(PolicyRule(
name="boost_chat",
condition=lambda e: e.get("type") == "chat",
action=lambda e, ctx: setattr(ctx, "priority_boost", 5) or ctx,
))
kernel.policy.add_rule(PolicyRule(
name="long_message_route",
condition=lambda e: len(e.get("message", "")) > 500,
action=lambda e, ctx: setattr(ctx, "route_override", "brain") or ctx,
))
kernel.policy.add_rule(PolicyRule(
name="route_memory",
condition=lambda e: e.get("type") == "memory",
action=lambda e, ctx: setattr(ctx, "route_override", "memory") or ctx,
))
kernel.policy.add_rule(PolicyRule(
name="route_tool",
condition=lambda e: e.get("type") == "tool",
action=lambda e, ctx: setattr(ctx, "route_override", "tool") or ctx,
))
kernel.policy.add_rule(PolicyRule(
name="route_voice",
condition=lambda e: e.get("type") == "voice",
action=lambda e, ctx: setattr(ctx, "route_override", "brain") or ctx,
))
chat = ChatExecutor()
memory = MemoryExecutor()
tool = ToolExecutor()
tool.register("echo", lambda p: {"echo": p.get("text", "")})
tool.register("now", lambda p: {"utc_ms": __import__("time").time() * 1000})
kernel.register_service("brain", chat)
kernel.register_service("voice", chat)
kernel.register_service("network", chat)
kernel.register_service("chat", chat)
kernel.register_service("memory", memory)
kernel.register_service("tool", tool)
return kernel
async def _run_task(event: dict, auth: dict | None):
kernel = build_kernel()
user_id = (auth or {}).get("user_id", "anonymous")
event.setdefault("user_id", user_id)
result = await kernel.handle(event)
return result
@app.get("/health")
async def health():
return {
"status": "ok",
"brain": "q-core-v13",
"openai": bool(os.getenv("OPENAI_API_KEY", "")),
"groq": bool(os.getenv("GROQ_API_KEY", "")),
"stripe": bool(os.getenv("STRIPE_SECRET_KEY", "")),
"queue": _queue.stats(),
}
@app.post("/brain")
async def brain(request: dict, auth: dict | None = Depends(optional_auth)):
return await _run_task(
{
"type": "chat",
"message": request.get("message", ""),
"context": request.get("context", []),
"model": request.get("model", "fast"),
},
auth,
)
@app.post("/v13/task")
async def v13_task(request: dict, auth: dict | None = Depends(optional_auth)):
task_type = request.get("type", "chat")
payload = request.get("payload") or {}
mode = request.get("mode", "sync")
if mode != "async":
return await _run_task({"type": task_type, **payload}, auth)
task_id = _queue.enqueue(task_type, {"auth_user": (auth or {}).get("user_id"), **payload})
return {"task_id": task_id, "status": "queued", "mode": "async"}
@app.get("/v13/task/{task_id}")
async def v13_task_status(task_id: str, auth: dict | None = Depends(optional_auth)):
row = _queue.get(task_id)
if row is None:
raise HTTPException(status_code=404, detail="task_not_found")
return row
@app.post("/v13/task/process")
async def v13_process_queue(auth: dict | None = Depends(optional_auth)):
processed = []
while True:
task = _queue.dequeue()
if task is None:
break
result = await _run_task(
{"type": task["type"], **task["payload"]}, auth
)
if result.get("error"):
_queue.fail(task["id"], result["error"])
else:
_queue.complete(task["id"], result)
processed.append({"id": task["id"], "result": result})
return {"processed": processed, "stats": _queue.stats()}
@app.get("/v13/queue/stats")
async def v13_queue_stats(auth: dict | None = Depends(optional_auth)):
return _queue.stats()
@app.post("/event")
async def handle_event(event: dict, auth: dict | None = Depends(optional_auth)):
return await _run_task(event, auth)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001) |