diff --git a/c/coli b/c/coli index 8c89caf9..b8b38a3b 100755 --- a/c/coli +++ b/c/coli @@ -159,8 +159,27 @@ def resource_request(a, env): [int(value) for value in gpu.split(",")]) return ram,ctx,devices,vram +def detect_slow_drive(model_path): + if os.environ.get("COLI_FORCE_NVME") == "1": return False + if os.environ.get("COLI_FORCE_HDD") == "1": return True + try: + if sys.platform == "win32": + out = subprocess.check_output(["powershell", "-NoProfile", "-Command", "Get-PhysicalDisk | Select-Object -ExpandProperty MediaType"], text=True, timeout=2).strip().lower() + return "hdd" in out + elif sys.platform.startswith("linux"): + out = subprocess.check_output(["lsblk", "-ndo", "ROTA", model_path], text=True, timeout=2).strip() + return out == "1" + except Exception: + pass + return False + def env_for(a): e = dict(os.environ, SNAP=a.model) + if detect_slow_drive(a.model): + print(f" {C.org}[AUTO-DETECT] Mechanical HDD detected. Enabling RAM-Aware Fallback Routing.{C.r}", file=sys.stderr) + e["COLI_HDD_MODE"] = "1" + e["AUTO_LATENCY_TARGET_MS"] = "500" + e["COLI_POLICY"]=a.policy if a.ram: e["RAM_GB"]=str(a.ram) if a.ngen: e["NGEN"]=str(a.ngen) diff --git a/c/glm.c b/c/glm.c index 8da3f553..60afb7bd 100644 --- a/c/glm.c +++ b/c/glm.c @@ -843,6 +843,10 @@ static float g_temp=-1; /* TEMP: temperatura di sampling sui TOKEN. <0 = auto ( static float g_nuc=0.95f;/* NUCLEUS: top-p sul vocabolario (default dal generation_config GLM-5.2) */ static int g_topk=0; /* TOPK=n -> usa n expert/token invece di config (ricerca: meno disco) */ static float g_topp=0; /* TOPP=p (0..1) -> top-p adattivo: tieni gli expert fino a peso cumulato p */ +static int g_hdd_mode=0; +static double g_auto_latency_target=0.0; +static double g_ewait_latency=0.0; +static uint64_t g_ewait_samples=0; /* CACHE_ROUTE (paper 2412.00099 max-rank): opt-in only. Keep true top-J always; * fill remaining slots preferring pin∪LRU experts ranked within top-M (or mass ROUTE_P). */ static int g_cache_route=0; @@ -2016,7 +2020,7 @@ static void attention_rows(Model *m, Layer *l, int layer, float *x, int S, int p for(int s=0;skv; int pos=positions?positions[s]:pos_base+s; - const float *qp=Q+(int64_t)s*H*qh+(int64_t)h*qh; + const float *qp=Q+(int64_t)s*H*qh+(int64_t)h*qh; /* [qk_nope | qk_rope] */ const float *qr=qp+c->qk_nope; int rbase=h*(c->qk_nope+vh); float qabs[512]; memset(qabs,0,kvl*sizeof(float)); @@ -2200,6 +2204,11 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int idx[chosen]=e; w[chosen]=rank_w[r]; chosen++; } } + if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target && chosen > 0) { + /* RAM-Aware Fallback: disk is too slow, STOP looking for uncached experts */ + Ksel = chosen; + } + /* Fill remainder from true ranking order. */ for(int r=0;rbv){bv=choice[e];best=e;} } + int eligible = 1; + if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target && kk > 0) { + eligible = expert_is_resident(m,layer,e); + } + if(!tk && eligible && choice[e]>bv){bv=choice[e];best=e;} } + if(best<0) break; idx[kk]=best; w[kk]=logit[best]; } + if(g_hdd_mode && g_auto_latency_target > 0 && g_ewait_latency > g_auto_latency_target) { + int c_k = 0; for(int kk=0;kk=0 && w[kk] > -1e20f) c_k++; + Ksel = c_k > 0 ? c_k : 1; /* Fallback to 1 if nothing */ + } + if(g_route_agree){ m->route_agree_hit+=(uint64_t)Ksel; m->route_agree_tot+=(uint64_t)Ksel; @@ -2398,10 +2417,21 @@ static void moe(Model *m, Layer *l, int layer, float *x, int S, float *out, int int eids[64]; for(int q=0;qt_edisk += now_s()-t0; /* dispatch only; real reads hide behind matmul */ - } else { double t0=now_s(); /* ORIGINALE: blocking parallel load */ - #pragma omp parallel for schedule(dynamic,1) - for(int q=0;qws[q],1); - m->t_edisk += now_s()-t0; } + } else { + double t0=now_s(); + for(int q=0;qt_edisk += read_ms/1000.0; + if(g_hdd_mode && nmiss > 0) { + double per_miss = read_ms / nmiss; + if(g_ewait_samples == 0) g_ewait_latency = per_miss; + else g_ewait_latency = g_ewait_latency * 0.8 + per_miss * 0.2; + g_ewait_samples++; + } + } } /* I/O ASINCRONO: readahead (WILLNEED) del blocco SUCCESSIVO mentre calcoliamo * questo — il kernel legge in background, le pread dopo trovano cache calda */ @@ -4763,6 +4793,9 @@ int main(int argc, char **argv){ if(g_mmap) fprintf(stderr,"[MMAP] expert = viste zero-copy nei file (page cache = cache)\n"); g_topk = getenv("TOPK")?atoi(getenv("TOPK")):0; g_topp = getenv("TOPP")?atof(getenv("TOPP")):0; + g_hdd_mode = getenv("COLI_HDD_MODE")?atoi(getenv("COLI_HDD_MODE")):0; + g_auto_latency_target = getenv("AUTO_LATENCY_TARGET_MS")?atof(getenv("AUTO_LATENCY_TARGET_MS")):0.0; + g_cache_route = getenv("CACHE_ROUTE")?atoi(getenv("CACHE_ROUTE")):0; g_route_j = getenv("ROUTE_J")?atoi(getenv("ROUTE_J")):2; g_route_m = getenv("ROUTE_M")?atoi(getenv("ROUTE_M")):12; diff --git a/c/openai_server.py b/c/openai_server.py index f31e8f7b..9005a1b7 100644 --- a/c/openai_server.py +++ b/c/openai_server.py @@ -58,6 +58,73 @@ def error_object(error): "param": error.param, "code": error.code}} +class MemoryManager: + def __init__(self): + self.base_dir = Path.home() / ".colibri" / "memory" + self.base_dir.mkdir(parents=True, exist_ok=True) + self.lock = threading.Lock() + + def _get_path(self, user): + safe_user = "".join(c for c in str(user) if c.isalnum() or c in "._-") + if not safe_user: return None + return self.base_dir / f"{safe_user}.txt" + + def search_and_inject(self, user, messages): + if not user or not messages: return + path = self._get_path(user) + if not path or not path.exists(): return + + last_user_msg = "" + for m in reversed(messages): + if m.get("role") == "user": + content = m.get("content", "") + if isinstance(content, str): last_user_msg = content + elif isinstance(content, list): last_user_msg = " ".join(p.get("text", "") for p in content if p.get("type") == "text") + break + + if not last_user_msg: return + words = [w.strip(".,!?\"'()[]{}:;") for w in last_user_msg.split()] + keywords = {w.lower() for w in words if len(w) > 4} + if not keywords: return + + matched_lines = [] + try: + with self.lock: + lines = path.read_text(encoding="utf-8").splitlines() + for line in lines: + line_lower = line.lower() + if any(kw in line_lower for kw in keywords): + matched_lines.append(line) + except Exception: + return + + if matched_lines: + # deduplicate + unique = list(dict.fromkeys(matched_lines)) + # take top 10 matches + unique = unique[-10:] + injection = "[LONG-TERM MEMORY]\n" + "\n".join(unique) + "\n[/LONG-TERM MEMORY]\n\n" + if messages[0].get("role") == "system": + messages[0]["content"] = injection + (messages[0].get("content") or "") + else: + messages.insert(0, {"role": "system", "content": injection}) + + def append_journal(self, user, user_msg, assistant_msg): + if not user or not user_msg or not assistant_msg: return + path = self._get_path(user) + if not path: return + try: + with self.lock: + with open(path, "a", encoding="utf-8") as f: + # Clean up newlines for a more compact PDF-like single-line search index + u_clean = user_msg.replace('\n', ' ').strip() + a_clean = assistant_msg.replace('\n', ' ').strip() + f.write(f"User: {u_clean} | Assistant: {a_clean}\n") + except Exception: + pass + +GLOBAL_MEMORY_MANAGER = MemoryManager() + class GenerationScheduler: """Bounded FIFO admission for the engine's independent KV contexts.""" @@ -826,7 +893,7 @@ def do_POST(self): except OSError: pass - def generation(self, body, prompt, request_id, chat): + def generation(self, body, prompt, request_id, chat, user_id=None, last_user_msg=None): # COLI_DEBUG tees the engine transaction to stderr: 1 = decoded output stream only, # 2 = both sides (rendered prompt + output). render_chat already folds prior turns and # tool results into `prompt`, so level 2 is the full conversation the engine saw. @@ -880,7 +947,12 @@ def generation(self, body, prompt, request_id, chat): choice = ({"index": 0, "message": {"role": "assistant", "content": text, "refusal": None}, "logprobs": None, "finish_reason": length_finish} if chat else {"index": 0, "text": text, "logprobs": None, "finish_reason": length_finish}) + + if user_id and last_user_msg: + GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, text) + self.send_json(200, {"id": completion_id, "object": object_name, "created": created, + "model": self.server.model_id, "choices": [choice], "usage": self.usage(stats)}, request_id, queue_headers) return @@ -976,7 +1048,12 @@ def emit_tools(chunk): lambda: not connected) if not sp["tool"] and sp["buf"]: emit(sp["buf"]) # no tool call happened: flush held tail + + if user_id and last_user_msg: + GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, "".join(raw)) + _content, calls = parse_tool_calls("".join(raw), tools) + for i, tc in enumerate(calls): event([{"index": 0, "delta": {"tool_calls": [{"index": i, "id": tc["id"], "type": "function", "function": {"name": tc["function"]["name"], @@ -984,14 +1061,19 @@ def emit_tools(chunk): "logprobs": None, "finish_reason": None}]) finish = "tool_calls" if calls else ("length" if stats["length_limited"] else "stop") else: + raw_text_acc = [] def emit_plain(chunk): + raw_text_acc.append(chunk) if dbg_echo: sys.stderr.write(chunk); sys.stderr.flush() emit(chunk) stats = self.server.engine.generate( prompt, maximum, temperature, top_p, emit_plain, cache_slot, lambda: not connected) + if user_id and last_user_msg: + GLOBAL_MEMORY_MANAGER.append_journal(user_id, last_user_msg, "".join(raw_text_acc)) finish = "length" if stats["length_limited"] else "stop" + ka_stop.set() # generation done: stop the keepalive pump ka_thread.join(timeout=2) final_choice = ({"index": 0, "delta": {}, "logprobs": None, "finish_reason": finish} @@ -1000,7 +1082,9 @@ def emit_plain(chunk): event([final_choice]) if include_usage: event([], self.usage(stats)) + if connected: + try: self.wfile.write(b"data: [DONE]\n\n") self.wfile.flush() @@ -1041,9 +1125,23 @@ def chat_completion(self, body, request_id): if not isinstance(enable_thinking, bool): raise APIError(400, "`enable_thinking` must be a boolean.", "enable_thinking") tools = body.get("tools") or body.get("functions") or None - prompt = render_chat(body.get("messages"), enable_thinking, reasoning_effort, tools, + + messages = body.get("messages", []) + user_id = body.get("user") + last_user_msg = "" + if user_id: + # Capture the last user message to append to the journal later + for m in reversed(messages): + if m.get("role") == "user": + c = m.get("content", "") + last_user_msg = c if isinstance(c, str) else " ".join(p.get("text", "") for p in c if p.get("type") == "text") + break + GLOBAL_MEMORY_MANAGER.search_and_inject(user_id, messages) + + prompt = render_chat(messages, enable_thinking, reasoning_effort, tools, body.get("tool_choice")) - self.generation(body, prompt, request_id, True) + self.generation(body, prompt, request_id, True, user_id=user_id, last_user_msg=last_user_msg) + def completion(self, body, request_id): prompt = body.get("prompt")