diff --git a/LICENSE b/LICENSE index 82584a9..817a86a 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2025 DevMind Contributors +Copyright (c) 2025 Synapse Contributors Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/server/app/__init__.py b/server/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/api/__init__.py b/server/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/api/router.py b/server/app/api/router.py new file mode 100644 index 0000000..daf10ad --- /dev/null +++ b/server/app/api/router.py @@ -0,0 +1,7 @@ +from fastapi import APIRouter + +from app.api.routes import chat, health + +router = APIRouter() +router.include_router(health.router) +router.include_router(chat.router) diff --git a/server/app/api/routes/__init__.py b/server/app/api/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/api/routes/chat.py b/server/app/api/routes/chat.py new file mode 100644 index 0000000..c5d535d --- /dev/null +++ b/server/app/api/routes/chat.py @@ -0,0 +1,36 @@ +from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import StreamingResponse + +from app.core.prompts import SYSTEM_PROMPT +from app.middleware.rate_limit import check_rate_limit +from app.models.chat import ChatRequest +from app.services.context import maybe_summarize, preprocess, trim_to_budget +from app.services.groq_client import get_groq_client +from app.services.streaming import stream_with_retry + +router = APIRouter() + + +@router.post("/chat", dependencies=[Depends(check_rate_limit)]) +async def chat(body: ChatRequest) -> StreamingResponse: + """Stream an LLM response for the given conversation history.""" + client = get_groq_client() + + try: + messages = preprocess(body.messages) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + messages = maybe_summarize(client, messages) + messages = trim_to_budget(messages) + + groq_messages = [{"role": "system", "content": SYSTEM_PROMPT}, *messages] + + return StreamingResponse( + stream_with_retry(client, groq_messages), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", # disable nginx buffering + }, + ) diff --git a/server/app/api/routes/health.py b/server/app/api/routes/health.py new file mode 100644 index 0000000..2d33cff --- /dev/null +++ b/server/app/api/routes/health.py @@ -0,0 +1,8 @@ +from fastapi import APIRouter + +router = APIRouter() + + +@router.get("/health") +async def health() -> dict[str, str]: + return {"status": "ok"} diff --git a/server/app/core/__init__.py b/server/app/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/core/config.py b/server/app/core/config.py new file mode 100644 index 0000000..5d064be --- /dev/null +++ b/server/app/core/config.py @@ -0,0 +1,32 @@ +from functools import lru_cache + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + extra="ignore", + ) + + groq_api_key: str + + # Model + groq_model: str = "llama-3.3-70b-versatile" + + max_context_tokens: int = 6_000 + summary_threshold: int = 20 + max_message_chars: int = 8_000 + rate_limit_requests: int = 20 + rate_limit_window_s: int = 60 + max_retries: int = 3 + retry_base_delay_s: float = 0.5 + + # CORS — set as JSON array in env: CORS_ORIGINS='["https://example.com"]' + cors_origins: list[str] = ["http://localhost:5173"] + + +@lru_cache +def get_settings() -> Settings: + return Settings() diff --git a/server/app/core/prompts.py b/server/app/core/prompts.py new file mode 100644 index 0000000..0f96771 --- /dev/null +++ b/server/app/core/prompts.py @@ -0,0 +1,11 @@ +SYSTEM_PROMPT = ( + "You are Synapse, a helpful and knowledgeable AI assistant for developers. " + "You excel at explaining code, debugging, architecture decisions, and technical concepts. " + "Format code examples with appropriate markdown code blocks." +) + +SUMMARIZE_PROMPT = ( + "Summarize the following conversation history as concisely as possible. " + "Preserve all key facts, decisions, code snippets, and context that would be " + "needed to continue the conversation. Output only the summary, no preamble." +) diff --git a/server/app/main.py b/server/app/main.py new file mode 100644 index 0000000..df18d72 --- /dev/null +++ b/server/app/main.py @@ -0,0 +1,39 @@ +""" +Synapse Backend — FastAPI application factory. +""" + +import logging + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from app.api.router import router +from app.core.config import get_settings + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + + +def create_app() -> FastAPI: + cfg = get_settings() + + app = FastAPI( + title="Synapse API", + version="2.0.0", + docs_url="/docs", + redoc_url="/redoc", + ) + + app.add_middleware( + CORSMiddleware, + allow_origins=cfg.cors_origins, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + app.include_router(router) + + return app + + +app = create_app() diff --git a/server/app/middleware/__init__.py b/server/app/middleware/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/middleware/rate_limit.py b/server/app/middleware/rate_limit.py new file mode 100644 index 0000000..ba70407 --- /dev/null +++ b/server/app/middleware/rate_limit.py @@ -0,0 +1,49 @@ +import time +from collections import deque + +from fastapi import HTTPException, Request + +from app.core.config import get_settings + +_windows: dict[str, deque[float]] = {} + + +def check_rate_limit(request: Request) -> None: + """ + Sliding-window rate limiter (per IP). + + Maintains a deque of request timestamps per IP. On each call: + 1. Evict timestamps older than the window from the left — O(k) where k + is the number of expired entries (amortised O(1) per request). + 2. Reject if the remaining count meets the limit. + 3. Append the current timestamp. + + A periodic sweep removes deques for IPs that have been idle longer than + one window, bounding memory to O(active IPs × window size). + """ + cfg = get_settings() + now = time.monotonic() + ip = request.client.host if request.client else "unknown" + window = _windows.setdefault(ip, deque()) + + while window and now - window[0] > cfg.rate_limit_window_s: + window.popleft() + + if len(window) >= cfg.rate_limit_requests: + raise HTTPException( + status_code=429, + detail=( + f"Rate limit exceeded: max {cfg.rate_limit_requests} requests " + f"per {cfg.rate_limit_window_s}s." + ), + ) + + window.append(now) + + if len(_windows) > 10_000: + stale = [ + k for k, v in _windows.items() + if not v or now - v[-1] > cfg.rate_limit_window_s + ] + for k in stale: + del _windows[k] diff --git a/server/app/models/__init__.py b/server/app/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/models/chat.py b/server/app/models/chat.py new file mode 100644 index 0000000..d561001 --- /dev/null +++ b/server/app/models/chat.py @@ -0,0 +1,24 @@ +from pydantic import BaseModel, field_validator + + +class Message(BaseModel): + role: str + content: str + + @field_validator("role") + @classmethod + def validate_role(cls, v: str) -> str: + if v not in {"user", "assistant", "system"}: + raise ValueError("role must be 'user', 'assistant', or 'system'") + return v + + +class ChatRequest(BaseModel): + messages: list[Message] + + @field_validator("messages") + @classmethod + def validate_messages(cls, v: list[Message]) -> list[Message]: + if not v: + raise ValueError("messages list cannot be empty") + return v diff --git a/server/app/services/__init__.py b/server/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/app/services/context.py b/server/app/services/context.py new file mode 100644 index 0000000..aaf3e38 --- /dev/null +++ b/server/app/services/context.py @@ -0,0 +1,127 @@ +""" +Context window management: token estimation, history trimming, and +conversation summarization. +""" + +import logging + +from groq import Groq + +from app.core.config import get_settings +from app.core.prompts import SUMMARIZE_PROMPT +from app.models.chat import Message + +logger = logging.getLogger(__name__) + +GroqMessage = dict[str, str] + + + +def estimate_tokens(text: str) -> int: + """ + O(n) approximation: 1 token ≈ 4 UTF-8 characters for English prose. + Accurate enough for budget decisions without a full tokenizer dependency. + """ + return max(1, len(text) // 4) + + +def preprocess(messages: list[Message]) -> list[GroqMessage]: + """ + Strip whitespace, enforce the per-message character cap, and drop empty + turns. Returns plain dicts ready for the Groq API. + + Raises ValueError if every message is empty after sanitization. + """ + cfg = get_settings() + result: list[GroqMessage] = [] + + for m in messages: + content = m.content.strip() + if not content: + continue + if len(content) > cfg.max_message_chars: + content = content[: cfg.max_message_chars] + "\n[Message truncated — input too long]" + result.append({"role": m.role, "content": content}) + + if not result: + raise ValueError("All messages were empty after sanitization") + + return result + + +def trim_to_budget(messages: list[GroqMessage]) -> list[GroqMessage]: + """ + Greedy reverse scan: walk messages newest → oldest, accumulating token cost + until MAX_CONTEXT_TOKENS is exhausted. The result is the largest suffix of + the conversation that fits within the budget. + + If messages were dropped, two framing turns are prepended so the model + knows the conversation was truncated and doesn't hallucinate earlier context. + """ + cfg = get_settings() + budget = cfg.max_context_tokens + kept: list[GroqMessage] = [] + + for msg in reversed(messages): + cost = estimate_tokens(msg["content"]) + if kept and budget - cost < 0: + break + budget -= cost + kept.append(msg) + + kept.reverse() + + if len(kept) < len(messages): + logger.info( + "Context trimmed: %d → %d messages (%d tokens remaining in budget)", + len(messages), + len(kept), + budget, + ) + kept = [ + {"role": "user", "content": "[Earlier messages omitted — context window full]"}, + {"role": "assistant", "content": "Understood. I'll work from the available context."}, + *kept, + ] + + return kept + + +def _summarize(client: Groq, messages: list[GroqMessage]) -> str: + """Non-streaming Groq call that returns a compact summary of older turns.""" + transcript = "\n".join(f"{m['role'].upper()}: {m['content']}" for m in messages) + response = client.chat.completions.create( + model=get_settings().groq_model, + messages=[ + {"role": "system", "content": SUMMARIZE_PROMPT}, + {"role": "user", "content": transcript}, + ], + stream=False, + temperature=0.3, + max_tokens=512, + ) + return response.choices[0].message.content or "" + + +def maybe_summarize(client: Groq, messages: list[GroqMessage]) -> list[GroqMessage]: + """ + When history exceeds SUMMARY_THRESHOLD messages, split at the midpoint, + summarize the older half, and replace it with a single assistant turn. + + This preserves the substance of earlier turns better than blind truncation + while keeping the payload sent to the model bounded. + """ + cfg = get_settings() + if len(messages) <= cfg.summary_threshold: + return messages + + split = len(messages) // 2 + old, recent = messages[:split], messages[split:] + + logger.info("Summarizing %d older messages (of %d total)", split, len(messages)) + summary = _summarize(client, old) + + return [ + {"role": "assistant", "content": f"[Summary of earlier conversation]\n{summary}"}, + *recent, + ] diff --git a/server/app/services/groq_client.py b/server/app/services/groq_client.py new file mode 100644 index 0000000..207d214 --- /dev/null +++ b/server/app/services/groq_client.py @@ -0,0 +1,13 @@ +from groq import Groq + +from app.core.config import get_settings + +_client: Groq | None = None + + +def get_groq_client() -> Groq: + """Lazy singleton — instantiated once on first request.""" + global _client + if _client is None: + _client = Groq(api_key=get_settings().groq_api_key) + return _client diff --git a/server/app/services/streaming.py b/server/app/services/streaming.py new file mode 100644 index 0000000..23da814 --- /dev/null +++ b/server/app/services/streaming.py @@ -0,0 +1,67 @@ +""" +SSE streaming with exponential-backoff retry. +""" + +import logging +import time +from collections.abc import Generator + +from groq import Groq + +from app.core.config import get_settings + +logger = logging.getLogger(__name__) + +GroqMessage = dict[str, str] + + +def stream_with_retry(client: Groq, messages: list[GroqMessage]) -> Generator[str, None, None]: + """ + Yield SSE-formatted chunks from Groq, retrying on transient failures with + exponential backoff. + + Retries are safe here because this generator is evaluated lazily inside + StreamingResponse — no data has reached the client when a connection to + Groq fails, so we can start the stream from scratch on each attempt. + + Backoff schedule (base=0.5s): attempt 1 → 0.5s, attempt 2 → 1s, attempt 3 → 2s. + On total failure a user-readable error event is emitted so the frontend can + surface it rather than receiving a silent broken stream. + """ + cfg = get_settings() + last_exc: Exception | None = None + + for attempt in range(cfg.max_retries): + try: + with client.chat.completions.create( + model=cfg.groq_model, + messages=messages, + stream=True, + temperature=0.7, + max_tokens=4096, + ) as stream: + for chunk in stream: + delta = chunk.choices[0].delta + if delta.content: + escaped = delta.content.replace("\n", "\\n") + yield f"data: {escaped}\n\n" + + yield "data: [DONE]\n\n" + return + + except Exception as exc: + last_exc = exc + if attempt < cfg.max_retries - 1: + delay = cfg.retry_base_delay_s * (2**attempt) + logger.warning( + "Groq request failed (attempt %d/%d), retrying in %.1fs — %s", + attempt + 1, + cfg.max_retries, + delay, + exc, + ) + time.sleep(delay) + + logger.error("All %d attempts failed: %s", cfg.max_retries, last_exc) + yield "data: [ERROR] The AI service is temporarily unavailable. Please try again.\n\n" + yield "data: [DONE]\n\n" diff --git a/server/main.py b/server/main.py deleted file mode 100644 index 43ecaaf..0000000 --- a/server/main.py +++ /dev/null @@ -1,105 +0,0 @@ -""" -Synapse Backend — FastAPI + Groq streaming chat API. -""" - -import os -from collections.abc import Generator - -from dotenv import load_dotenv -from fastapi import FastAPI, HTTPException -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import StreamingResponse -from groq import Groq -from pydantic import BaseModel, field_validator - -load_dotenv() - -app = FastAPI(title="Synapse API", version="1.0.0") - -app.add_middleware( - CORSMiddleware, - allow_origins=["http://localhost:5173"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - -MODEL = "llama-3.3-70b-versatile" -SYSTEM_PROMPT = ( - "You are Synapse, a helpful and knowledgeable AI assistant for developers. " - "You excel at explaining code, debugging, architecture decisions, and technical concepts. " - "Format code examples with appropriate markdown code blocks." -) - - -class Message(BaseModel): - role: str - content: str - - @field_validator("role") - @classmethod - def validate_role(cls, v: str) -> str: - if v not in {"user", "assistant", "system"}: - raise ValueError("role must be 'user', 'assistant', or 'system'") - return v - - -class ChatRequest(BaseModel): - messages: list[Message] - - @field_validator("messages") - @classmethod - def validate_messages(cls, v: list[Message]) -> list[Message]: - if not v: - raise ValueError("messages list cannot be empty") - return v - - -def _get_groq_client() -> Groq: - api_key = os.getenv("GROQ_API_KEY") - if not api_key: - raise HTTPException(status_code=500, detail="GROQ_API_KEY is not configured") - return Groq(api_key=api_key) - - -def _stream_chat(messages: list[Message]) -> Generator[str, None, None]: - """Yield SSE-formatted chunks from the Groq streaming API.""" - client = _get_groq_client() - - groq_messages = [{"role": "system", "content": SYSTEM_PROMPT}] + [ - {"role": m.role, "content": m.content} for m in messages - ] - - with client.chat.completions.create( - model=MODEL, - messages=groq_messages, - stream=True, - temperature=0.7, - max_tokens=4096, - ) as stream: - for chunk in stream: - delta = chunk.choices[0].delta - if delta.content: - # Escape newlines so they don't break the SSE frame - escaped = delta.content.replace("\n", "\\n") - yield f"data: {escaped}\n\n" - - yield "data: [DONE]\n\n" - - -@app.post("/chat") -async def chat(request: ChatRequest) -> StreamingResponse: - """Stream an LLM response for the given conversation history.""" - return StreamingResponse( - _stream_chat(request.messages), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "X-Accel-Buffering": "no", # disable nginx buffering - }, - ) - - -@app.get("/health") -async def health() -> dict[str, str]: - return {"status": "ok"} diff --git a/server/requirements.txt b/server/requirements.txt index 3f2ffcf..bed9ed2 100644 --- a/server/requirements.txt +++ b/server/requirements.txt @@ -1,5 +1,5 @@ fastapi>=0.115.0 uvicorn[standard]>=0.32.0 groq>=0.13.0 -python-dotenv>=1.0.0 pydantic>=2.9.0 +pydantic-settings>=2.0.0