| |
| """PALIMPSESTE — FastAPI web server with cognitive metrics + streaming. |
| |
| Endpoints: |
| GET / — serves the React production build (or fallback HTML) |
| GET /health — health check |
| GET /stats — model statistics |
| GET /metrics — rich metrics for dashboard visualization |
| POST /chat — send message, get response + confidence + source + explanation |
| POST /chat/stream — SSE streaming response with token-level data |
| POST /teach — teach a new Q/A pair at runtime (O(1)) |
| GET /memory/samples — recent memory traces for visualization |
| GET /conversation — current conversation transcript |
| |
| Usage: |
| pip install fastapi uvicorn |
| python examples/api_server.py --model thefinalboss/palimpseste-max --port 3332 |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| import time |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
|
|
| from palimseste.hf import HFPalimpsesteLM |
| from palimseste.chat import Conversation |
| from palimseste.cognitive import CognitiveAgent, CognitiveResponse |
|
|
| try: |
| from fastapi import FastAPI |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse |
| from pydantic import BaseModel |
| import uvicorn |
| HAS_FASTAPI = True |
| except ImportError: |
| HAS_FASTAPI = False |
|
|
|
|
| |
| class ChatRequest(BaseModel): |
| message: str |
| temperature: float | None = 0.0 |
| max_tokens: int = 200 |
|
|
|
|
| class TeachRequest(BaseModel): |
| question: str |
| answer: str |
|
|
|
|
| class LearnTextRequest(BaseModel): |
| text: str |
| tag: str | None = None |
|
|
|
|
| class DreamRequest(BaseModel): |
| cycles: int = 3 |
| replay_batch: int = 200 |
|
|
|
|
| def _chain_to_dict(chain): |
| if chain is None: |
| return None |
| return { |
| "success": chain.success, |
| "answer": chain.answer, |
| "original_question": chain.original_question, |
| "n_hops": chain.n_hops, |
| "steps": [ |
| { |
| "sub_question": s.sub_question, |
| "sub_answer": s.sub_answer, |
| "resolved_question": s.resolved_question, |
| } |
| for s in chain.steps |
| ], |
| } |
|
|
|
|
| def _cog_response_to_dict(resp: CognitiveResponse, elapsed_ms: float) -> dict: |
| return { |
| "response": resp.text, |
| "confidence": round(resp.confidence, 4), |
| "source": resp.source, |
| "explanation": resp.explanation, |
| "chain": _chain_to_dict(resp.chain), |
| "corrected_answer": resp.corrected_answer, |
| "elapsed_ms": round(elapsed_ms, 1), |
| } |
|
|
|
|
| |
| FALLBACK_HTML = """<!DOCTYPE html> |
| <html> |
| <head><title>PALIMPSESTE</title></head> |
| <body style="font-family:sans-serif;background:#0a0a12;color:#e0e0e0;padding:40px"> |
| <h1>PALIMPSESTE API</h1> |
| <p>API is running. Endpoints: /health /stats /metrics /chat /teach /memory/samples</p> |
| </body></html>""" |
|
|
|
|
| def create_app(lm: HFPalimpsesteLM, corpus_pairs=None) -> "FastAPI": |
| if not HAS_FASTAPI: |
| raise ImportError("FastAPI not installed. Run: pip install fastapi uvicorn") |
|
|
| app = FastAPI(title="PALIMPSESTE", description="Hypervectorial Cortex API") |
|
|
| |
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| conv = Conversation(model=lm, fuzzy_threshold=0.72, learn_live=True) |
| if corpus_pairs: |
| conv.register_questions(corpus_pairs) |
| agent = CognitiveAgent(conv=conv) |
|
|
| |
| from palimseste.cortex import InstantExpert, Dreamer, Composer |
| from palimseste.reasoning import Reasoner |
| expert = InstantExpert(lm=lm) |
| dreamer = Dreamer(mem=lm.mem, phi=lm.phi) |
| reasoner = Reasoner(conv=conv) |
| composer = Composer(reasoner=reasoner) |
|
|
| |
| from palimseste.cortex import ( |
| MultimodalFusion, MetaLearner, DualMemory, Analogizer, |
| ) |
| fusion = MultimodalFusion(mem=lm.mem, encoder=lm.encoder) |
| meta_learner = MetaLearner(mem=lm.mem, phi=lm.phi) |
| dual_mem = DualMemory(mem=lm.mem, encoder=lm.encoder) |
|
|
| |
| react_build = Path(__file__).resolve().parent.parent / "web" / "dist" |
| if react_build.exists(): |
| from fastapi.staticfiles import StaticFiles |
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def root(): |
| index = react_build / "index.html" |
| return HTMLResponse(index.read_text(encoding="utf-8")) |
|
|
| app.mount("/assets", StaticFiles(directory=str(react_build / "assets")), name="assets") |
| else: |
| @app.get("/", response_class=HTMLResponse) |
| async def root(): |
| return FALLBACK_HTML |
|
|
| @app.get("/health") |
| async def health(): |
| return {"status": "ok", "D": lm.config.D, "|M|": len(lm.mem)} |
|
|
| @app.get("/stats") |
| async def stats(): |
| s = lm.stats() |
| s["turns"] = conv.turn_count |
| s["n_corrections"] = agent.n_corrections |
| return JSONResponse(s) |
|
|
| @app.get("/metrics") |
| async def metrics(): |
| """Rich metrics for the dashboard.""" |
| mem_stats = lm.mem.stats() |
| s = lm.stats() |
| return JSONResponse({ |
| "model": { |
| "D": s["D"], |
| "n_traces": s["n_traces"], |
| "n_meta": s["n_meta"], |
| "vocab_size": s["vocab_size"], |
| "context_window": s["context_window"], |
| "theoretical_capacity_log2": s["theoretical_capacity_log2"], |
| }, |
| "memory": { |
| "mean_weight": round(mem_stats.mean_weight, 4), |
| "min_weight": round(mem_stats.min_weight, 4), |
| "lsh_size": mem_stats.lsh_size, |
| }, |
| "conversation": { |
| "turns": conv.turn_count, |
| "n_corrections": agent.n_corrections, |
| "known_questions": len(conv._known_questions), |
| }, |
| "long_context": lm.long_context_stats(), |
| "config": { |
| "kernel_radius": lm.config.kernel_radius, |
| "kernel_min_weight": lm.config.kernel_min_weight, |
| "temperature": lm.config.temperature, |
| "model_type": lm.config.model_type, |
| }, |
| }) |
|
|
| @app.get("/memory/samples") |
| async def memory_samples(limit: int = 50): |
| """Recent memory traces for visualization.""" |
| mem = lm.mem |
| total = len(mem) |
| n = min(limit, total) |
| if n == 0: |
| return JSONResponse({"traces": [], "total": 0}) |
| |
| traces = [] |
| for i in range(total - n, total): |
| t = mem._traces[i] |
| traces.append({ |
| "id": t.id, |
| "weight": round(t.weight, 4), |
| "meta": t.meta, |
| "tag": t.tag or "", |
| "age": total - t.id, |
| }) |
| return JSONResponse({"traces": traces, "total": total}) |
|
|
| @app.get("/conversation") |
| async def conversation(): |
| """Current conversation transcript.""" |
| turns = [] |
| for turn in conv.history: |
| turns.append({"role": turn.role, "text": turn.text}) |
| return JSONResponse({"turns": turns, "count": conv.turn_count}) |
|
|
| @app.post("/chat") |
| async def chat(req: ChatRequest): |
| t0 = time.perf_counter() |
| resp = agent.respond(req.message, max_new_tokens=req.max_tokens, |
| temperature=req.temperature, seed=None) |
| dt = (time.perf_counter() - t0) * 1000 |
| return JSONResponse(_cog_response_to_dict(resp, dt)) |
|
|
| @app.post("/chat/stream") |
| async def chat_stream(req: ChatRequest): |
| """SSE streaming with cognitive metadata.""" |
| t0 = time.perf_counter() |
|
|
| def generate(): |
| |
| resp = agent.respond(req.message, max_new_tokens=req.max_tokens, |
| temperature=req.temperature, seed=None) |
| dt = (time.perf_counter() - t0) * 1000 |
|
|
| |
| tokens = resp.text.split() |
| for i, token in enumerate(tokens): |
| chunk = { |
| "type": "token", |
| "text": token + (" " if i < len(tokens) - 1 else ""), |
| "index": i, |
| } |
| yield f"data: {json.dumps(chunk)}\n\n" |
| time.sleep(0.02) |
|
|
| |
| meta = _cog_response_to_dict(resp, dt) |
| meta["type"] = "metadata" |
| yield f"data: {json.dumps(meta)}\n\n" |
|
|
| return StreamingResponse(generate(), media_type="text/event-stream") |
|
|
| @app.post("/teach") |
| async def teach(req: TeachRequest): |
| n = len(lm.mem) |
| resp = agent.teach(req.question, req.answer) |
| return JSONResponse({ |
| "success": True, |
| "message": resp.text, |
| "tokens_written": len(lm.mem) - n, |
| "explanation": resp.explanation, |
| }) |
|
|
| |
|
|
| @app.post("/learn-text") |
| async def learn_text(req: LearnTextRequest): |
| """Feature 1: Instant Expertise — ingest a document, become expert.""" |
| n_before = len(lm.mem) |
| result = expert.learn_from_text(req.text, document_tag=req.tag, verbose=False) |
| return JSONResponse({ |
| "success": True, |
| "n_tokens": result.n_tokens, |
| "n_facts": result.n_facts, |
| "n_seconds": round(result.n_seconds, 2), |
| "document_tag": result.document_tag, |
| "facts": [{"q": q, "a": a} for q, a in result.facts[:20]], |
| "tokens_written": len(lm.mem) - n_before, |
| }) |
|
|
| @app.post("/dream") |
| async def dream(req: DreamRequest): |
| """Feature 2: Dream Consolidation — get smarter while idle.""" |
| n_before = dreamer.n_concepts |
| result = dreamer.dream(n_cycles=req.cycles, replay_batch=req.replay_batch) |
| return JSONResponse({ |
| "success": True, |
| "n_concepts_promoted": result.n_concepts_promoted, |
| "n_concepts_extracted": result.n_concepts_extracted, |
| "n_total_concepts": dreamer.n_concepts, |
| "n_new_concepts": dreamer.n_concepts - n_before, |
| "n_seconds": round(result.n_seconds, 2), |
| "concept_labels": result.concept_labels[:20], |
| "new_connections": result.new_connections[:10], |
| }) |
|
|
| @app.post("/reason") |
| async def reason(req: ChatRequest): |
| """Feature 3: Compositional Reasoning — decompose and resolve.""" |
| result = composer.reason(req.message) |
| return JSONResponse({ |
| "answer": result.answer, |
| "success": result.success, |
| "n_decompositions": result.n_decompositions, |
| "n_hops": result.n_hops, |
| "n_seconds": round(result.n_seconds, 3), |
| "steps": [ |
| { |
| "type": s.step_type, |
| "question": s.sub_question, |
| "answer": s.sub_answer, |
| "confidence": s.confidence, |
| } |
| for s in result.steps |
| ], |
| }) |
|
|
| @app.get("/cortex/status") |
| async def cortex_status(): |
| """Status of all 7 cortex features.""" |
| return JSONResponse({ |
| "expertise": { |
| "n_documents": expert.n_documents, |
| "documents": list(expert.documents().keys()), |
| }, |
| "dreamer": { |
| "n_concepts": dreamer.n_concepts, |
| "n_consolidated": dreamer.consolidator.n_promoted, |
| "n_abstracted": dreamer.abstraction.n_concepts, |
| }, |
| "multimodal": { |
| "n_bindings": fusion.n_bindings, |
| }, |
| "meta_learning": { |
| "n_adaptations": meta_learner.n_adaptations, |
| "current_config": meta_learner.current_config, |
| }, |
| "dual_memory": { |
| "n_episodic": dual_mem.n_episodic, |
| "n_semantic": dual_mem.n_semantic, |
| "total": dual_mem.total, |
| }, |
| }) |
|
|
| |
|
|
| @app.post("/adapt") |
| async def adapt(domain: str | None = None): |
| """Feature 5: Meta-learning — self-tune kernel parameters.""" |
| result = meta_learner.adapt(domain=domain) |
| return JSONResponse({ |
| "accepted": result.accepted, |
| "param_changed": result.param_changed, |
| "energy_before": round(result.energy_before, 4), |
| "energy_after": round(result.energy_after, 4), |
| "rationale": result.rationale, |
| "current_config": meta_learner.current_config, |
| }) |
|
|
| @app.post("/memory/episodic") |
| async def store_episodic(content: str, tag: str = ""): |
| """Feature 6: Store an episodic memory.""" |
| record = dual_mem.store_episodic(content, tag=tag) |
| return JSONResponse({ |
| "success": True, |
| "type": record.memory_type, |
| "n_episodic": dual_mem.n_episodic, |
| "n_semantic": dual_mem.n_semantic, |
| }) |
|
|
| @app.post("/memory/semantic") |
| async def store_semantic(content: str, tag: str = ""): |
| """Feature 6: Store a semantic memory (persistent fact).""" |
| record = dual_mem.store_semantic(content, tag=tag) |
| return JSONResponse({ |
| "success": True, |
| "type": record.memory_type, |
| "n_episodic": dual_mem.n_episodic, |
| "n_semantic": dual_mem.n_semantic, |
| }) |
|
|
| @app.post("/memory/recall") |
| async def recall_memory(query: str, top_k: int = 5): |
| """Feature 6: Recall memories by query.""" |
| results = dual_mem.recall(query, top_k=top_k) |
| return JSONResponse({ |
| "results": [ |
| { |
| "content": r.content, |
| "type": r.memory_type, |
| "tag": r.tag, |
| "score": round(s, 4), |
| "timestamp": r.timestamp, |
| } |
| for r, s in results |
| ], |
| "n_results": len(results), |
| }) |
|
|
| return app |
|
|
|
|
| def main(): |
| if not HAS_FASTAPI: |
| print("FastAPI not installed. Run: pip install fastapi uvicorn") |
| sys.exit(1) |
|
|
| parser = argparse.ArgumentParser(description="PALIMPSESTE FastAPI server") |
| parser.add_argument("--model", "-m", type=str, required=True, |
| help="path to model dir or HF Hub repo id") |
| parser.add_argument("--port", "-p", type=int, default=3332) |
| parser.add_argument("--host", type=str, default="0.0.0.0") |
| args = parser.parse_args() |
|
|
| print(f"loading model from {args.model} ...", file=sys.stderr) |
| try: |
| lm = HFPalimpsesteLM.from_pretrained(args.model) |
| except Exception as e: |
| print(f"Failed to load '{args.model}': {e}", file=sys.stderr) |
| |
| local = Path(__file__).resolve().parent.parent / "palimpseste-max" |
| if local.exists(): |
| print(f"Falling back to local model at {local} ...", file=sys.stderr) |
| lm = HFPalimpsesteLM.from_pretrained(str(local)) |
| else: |
| raise |
| print(f"loaded: D={lm.config.D:,} |M|={len(lm.mem):,}", file=sys.stderr) |
|
|
| corpus_pairs = None |
| try: |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from corpus_chat import get_corpus |
| corpus_pairs = get_corpus() |
| print(f"registered {len(corpus_pairs)} Q/A pairs for fuzzy matching", |
| file=sys.stderr) |
| except ImportError: |
| pass |
|
|
| app = create_app(lm, corpus_pairs) |
| print(f"\nPALIMPSESTE API server starting on http://{args.host}:{args.port}", |
| file=sys.stderr) |
| print(f"Open http://localhost:{args.port} in your browser\n", file=sys.stderr) |
| uvicorn.run(app, host=args.host, port=args.port, log_level="warning") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
