From 115a2f9c91e709818dfcc21dec900e5bfa745091 Mon Sep 17 00:00:00 2001 From: Adil Date: Thu, 6 Aug 2026 09:47:11 +0500 Subject: [PATCH] =?UTF-8?q?hive:=20ollama-worker=20=E2=80=94=20WorkerBee?= =?UTF-8?q?=20that=20calls=20Ollama's=20HTTP=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sends prompts to Ollama's /api/chat endpoint and streams the response back over thrum. Run any local LLM Ollama supports. Also modifies serve.rs (nest-common) to pass through chi events directly instead of requiring claude's stream-json format, so workers that emit thrum events work natively. Ollama-worker registers as a worker bee with configurable models via OLLAMA_WORKER_MODEL / OLLAMA_WORKER_MODELS env vars. --- Cargo.toml | 1 + hives/common/src/serve.rs | 13 ++ hives/ollama-worker/Cargo.toml | 25 ++++ hives/ollama-worker/Orchfile | 5 + hives/ollama-worker/src/lib.rs | 222 ++++++++++++++++++++++++++++++++ hives/ollama-worker/src/main.rs | 49 +++++++ 6 files changed, 315 insertions(+) create mode 100644 hives/ollama-worker/Cargo.toml create mode 100644 hives/ollama-worker/Orchfile create mode 100644 hives/ollama-worker/src/lib.rs create mode 100644 hives/ollama-worker/src/main.rs diff --git a/Cargo.toml b/Cargo.toml index 66e4563..aed08f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ members = [ "hives/paid-oracle", "hives/gsm-modem", "hives/ollama-server", + "hives/ollama-worker", "hives/bp7", "thehum", ] diff --git a/hives/common/src/serve.rs b/hives/common/src/serve.rs index 70c2a2b..7f4939d 100644 --- a/hives/common/src/serve.rs +++ b/hives/common/src/serve.rs @@ -538,6 +538,19 @@ impl WireListener { async fn forward_raw(&self, value: Value) { self.last_touched.store(now_ms(), Ordering::SeqCst); + + // If the event is already a thrum chi event (has a "chi" field), + // pass it through directly. This lets workers like ollama-worker + // emit thrum events without going through claude's stream-json. + if value.get("chi").and_then(Value::as_str).is_some() { + let mut v = value.clone(); + if let Some(obj) = v.as_object_mut() { + obj.insert("sid".into(), Value::String(self.sid.clone())); + } + self.send(v).await; + return; + } + // claude emits stream-json events. The relevant chunk events // arrive wrapped as `{"type":"stream_event","event":{...inner...}}`; // unwrap to inspect the inner type. Mirrors the dispatch diff --git a/hives/ollama-worker/Cargo.toml b/hives/ollama-worker/Cargo.toml new file mode 100644 index 0000000..d03b4aa --- /dev/null +++ b/hives/ollama-worker/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "ollama-worker" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "WorkerBee that calls Ollama's API directly — run local LLMs without claude." + +[[bin]] +name = "ollama-worker" +path = "src/main.rs" + +[dependencies] +hum-paths = { path = "../../hum-paths" } +nest = { path = "../../nest" } +nest-common = { path = "../common" } +ids = { path = "../../ids" } +tokio = { workspace = true, features = ["full"] } +tokio-util = "0.7" +serde_json = { workspace = true } +serde = { workspace = true } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream"] } +anyhow = { workspace = true } +async-trait = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/hives/ollama-worker/Orchfile b/hives/ollama-worker/Orchfile new file mode 100644 index 0000000..9c63728 --- /dev/null +++ b/hives/ollama-worker/Orchfile @@ -0,0 +1,5 @@ +SERVICE ollama-worker +RUN ${HOME}/.local/bin/ollama-worker +ENV OLLAMA_WORKER_MODEL=smollm:1.7b +ENV OLLAMA_WORKER_MODELS=smollm:1.7b +RESTART always diff --git a/hives/ollama-worker/src/lib.rs b/hives/ollama-worker/src/lib.rs new file mode 100644 index 0000000..de732cb --- /dev/null +++ b/hives/ollama-worker/src/lib.rs @@ -0,0 +1,222 @@ +//! ollama-worker — WorkerBee that calls Ollama's HTTP API. +//! +//! Sends prompts to Ollama's `/api/chat` endpoint and streams the +//! response back over thrum. Run any local LLM Ollama supports. +//! +//! Environment: +//! OLLAMA_WORKER_URL — Ollama API base URL (default: http://127.0.0.1:11434) +//! OLLAMA_WORKER_MODEL — default model (default: smollm:1.7b) +//! OLLAMA_WORKER_CTX — context length (default: 8192) + +use std::sync::Arc; + +use anyhow::Result; +use async_trait::async_trait; +use serde::Serialize; +use serde_json::{json, Value}; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use nest::{Cell, Egg, Propensity, WorkerBee}; + +/// Ollama chat request body — the exact shape Ollama's API expects. +#[derive(Debug, Clone, Serialize)] +struct OllamaRequest { + model: String, + messages: Vec, + stream: bool, + options: Option, +} + +#[derive(Debug, Clone, Serialize)] +struct OllamaMessage { + role: String, + content: String, +} + +#[derive(Debug, Clone, Serialize)] +struct OllamaOptions { + num_ctx: Option, + temperature: Option, + top_p: Option, + top_k: Option, + seed: Option, + stop: Option>, +} + +pub struct OllamaWorker { + url: String, + default_model: String, + default_ctx: u64, +} + +impl Default for OllamaWorker { + fn default() -> Self { + Self { + url: std::env::var("OLLAMA_WORKER_URL") + .unwrap_or_else(|_| "http://127.0.0.1:11434".into()), + default_model: std::env::var("OLLAMA_WORKER_MODEL") + .unwrap_or_else(|_| "smollm:1.7b".into()), + default_ctx: std::env::var("OLLAMA_WORKER_CTX") + .ok().and_then(|s| s.parse::().ok()) + .unwrap_or(8192), + } + } +} + +#[async_trait] +impl WorkerBee for OllamaWorker { + fn ephemeral(&self) -> bool { false } + + fn propensity(&self) -> Propensity { + // Ollama API is stateless per call — every request carries full history. + // The cell lives for one streaming response; humd can reuse it + // for follow-up turns in the same sid. + Propensity::StatefulSession + } + + async fn raise(&self, egg: Egg) -> Result { + let model = if egg.model_id.is_empty() { + self.default_model.clone() + } else { + egg.model_id.clone() + }; + + info!(model = %model, cwd = %egg.cwd, "ollama-worker.raise"); + let client = reqwest::Client::new(); + let url = self.url.clone(); + let default_ctx = self.default_ctx; + + let (tx_feed, mut rx_feed) = tokio::sync::mpsc::channel::(64); + let (tx_mmm, rx_mmm) = tokio::sync::mpsc::channel::(256); + let silence = tokio_util::sync::CancellationToken::new(); + let silence_clone = silence.clone(); + + let tx_mmm_wire = tx_mmm.clone(); + tokio::spawn(async move { + let prompt = match rx_feed.recv().await { + Some(p) => p, + None => { + warn!("ollama-worker: feed closed before first prompt"); + return; + } + }; + + let mut msgs = Vec::new(); + if let Some(sp) = &egg.system_prompt { + msgs.push(OllamaMessage { + role: "system".to_string(), + content: sp.clone(), + }); + } + msgs.push(OllamaMessage { + role: "user".to_string(), + content: prompt, + }); + + let request = OllamaRequest { + model: model.clone(), + messages: msgs, + stream: true, + options: Some(OllamaOptions { + num_ctx: Some(default_ctx), + temperature: None, + top_p: None, + top_k: None, + seed: None, + stop: None, + }), + }; + + let req_body = serde_json::to_string(&request).unwrap(); + let req = client + .post(format!("{}/api/chat", url)) + .header("Content-Type", "application/json") + .body(req_body); + + match req.send().await { + Ok(resp) => { + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + let err = json!({ + "chi": "error", + "code": "ollama_api_error", + "message": format!("Ollama returned {status}: {body}"), + }); + let _ = tx_mmm_wire.send(err).await; + return; + } + + let lines = resp.text().await.unwrap_or_default(); + + for line in lines.lines() { + if line.trim().is_empty() { continue; } + if silence_clone.is_cancelled() { break; } + + match serde_json::from_str::(line) { + Ok(chunk) => { + let done = chunk.get("done").and_then(|d| d.as_bool()).unwrap_or(false); + let content = chunk.get("message") + .and_then(|m| m.get("content")) + .and_then(|c| c.as_str()) + .unwrap_or(""); + + if !content.is_empty() { + let delta = json!({ + "chi": "chunk", + "chunkType": "text_delta", + "delta": content, + }); + let _ = tx_mmm_wire.send(delta).await; + } + + if done { + let eval_count = chunk.get("eval_count") + .and_then(|c| c.as_u64()) + .unwrap_or(0); + let prompt_eval_count = chunk.get("prompt_eval_count") + .and_then(|c| c.as_u64()) + .unwrap_or(0); + + let finish = json!({ + "chi": "finish", + "finishReason": "stop", + "usage": { + "input_tokens": prompt_eval_count, + "output_tokens": eval_count, + }, + }); + let _ = tx_mmm_wire.send(finish).await; + return; + } + } + Err(e) => { + warn!(err = %e, line = line.chars().take(200).collect::(), "ollama-worker.parse"); + } + } + } + } + Err(e) => { + let err = json!({ + "chi": "error", + "code": "ollama_connection_error", + "message": format!("Failed to reach Ollama at {url}: {e}"), + }); + let _ = tx_mmm_wire.send(err).await; + } + } + }); + + let (_tx_exit, rx_exit) = tokio::sync::oneshot::channel::(); + + Ok(Cell { + mark: None, + feed: tx_feed, + mmm: Arc::new(Mutex::new(rx_mmm)), + emerged: rx_exit, + ephemeral: false, + silence, + }) + } +} diff --git a/hives/ollama-worker/src/main.rs b/hives/ollama-worker/src/main.rs new file mode 100644 index 0000000..3b5e9f8 --- /dev/null +++ b/hives/ollama-worker/src/main.rs @@ -0,0 +1,49 @@ +//! ollama-worker — standalone worker-bee process. +//! +//! Registers with humd as `bee:["worker"]`, advertises models from +//! `OLLAMA_WORKER_MODELS`, and routes prompts to Ollama's API. +//! +//! Environment: +//! OLLAMA_WORKER_URL — Ollama API base URL (default: http://127.0.0.1:11434) +//! OLLAMA_WORKER_MODEL — default model (default: llama3.2) +//! OLLAMA_WORKER_MODELS — comma-separated model list (default: OLLAMA_WORKER_MODEL) +//! OLLAMA_WORKER_CTX — context length (default: 8192) + +use std::sync::Arc; + +use anyhow::Result; +use nest_common::{serve_worker, HiveAdvert}; + +use lib::OllamaWorker; +mod lib; + +#[tokio::main] +async fn main() -> Result<()> { + hum_paths::init(); + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + + let default_model = std::env::var("OLLAMA_WORKER_MODEL") + .unwrap_or_else(|_| "smollm:1.7b".into()); + let models_env = std::env::var("OLLAMA_WORKER_MODELS") + .unwrap_or_else(|_| default_model.clone()); + let models: Vec = models_env + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + + let worker = Arc::new(OllamaWorker::default()); + let advert = HiveAdvert { + hive: "ollama-worker".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + models, + source: Some("https://github.com/adiled/hum/tree/main/hives/ollama-worker".to_string()), + }; + + serve_worker(worker, advert).await +}