diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 34bf835..4e3b6ea 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,6 +1,9 @@ name: CI -on: [push, pull_request] +on: + push: + branches: [main] + pull_request: permissions: contents: read diff --git a/Cargo.lock b/Cargo.lock index 6a6b4db..1e7ddee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1042,7 +1042,7 @@ checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" [[package]] name = "openai-api-dispatch" -version = "0.1.0" +version = "0.1.1" dependencies = [ "anyhow", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 35881fd..6da7018 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "openai-api-dispatch" -version = "0.1.0" +version = "0.1.1" authors = ["Victor Lopez "] description = "OpenAI-compatible chat requests through memory or NATS queues." edition = "2024" diff --git a/README.md b/README.md index 126bce4..236cfa3 100644 --- a/README.md +++ b/README.md @@ -59,21 +59,32 @@ Environment settings are read when constructing producers, workers, and executor | Variable | Default | | --- | --- | | `OPENAI_API_NATS_URL` | `nats://localhost:4222` | +| `OPENAI_API_NATS_WORKERS_GROUP` | `task_workers` | | `OPENAI_API_NATS_PREFIX` | `openai-api-queue/` | | `OPENAI_API_URL` | `http://127.0.0.1:8000/v1` | | `OPENAI_API_DEFAULT_MODEL` | Unset; required by NATS workers | -- A task's explicit model overrides the default and selects its NATS subject. Prefix and model are concatenated verbatim. Workers share the `task_workers` queue group; this uses Core NATS without persistence or task retries. Constructors do not wait for server confirmation of subscriptions, so startup can race with publishing. -- Each worker handles one task at a time. Queue/API errors stop its loop without an error reply; supervise spawned workers. `send_and_wait` limits only the reply wait, not submission, and expiry does not cancel execution. Check `response.success` even when the call returns `Ok`. -- Only non-streaming chat is implemented. The current prompt is sent as a user message; system entries in input history are ignored (use `with_system`). Returned history currently duplicates the latest user prompt before the assistant reply. Schemas request strict JSON output without local validation. Keep `max_tokens` within `u32`; it is cast unchecked. `payload` is caller metadata, not model input. +- A task's explicit model overrides the default and selects its NATS subject. +- Constructors do not wait for server confirmation of subscriptions, so startup can race with publishing. +- Each worker handles one task at a time. +- Queue/API errors stop its loop without an error reply; supervise spawned workers. +- `send_and_wait` limits only the reply wait, not submission, and expiry does not cancel execution. +- Check `response.success` even when the call returns `Ok`. +- Only non-streaming chat is implemented. +- The current prompt is sent as a user message; system entries in input history are ignored (use `with_system`). +- Schemas request strict JSON output without local validation. +- `payload` is caller metadata, not model input. +- `no_std` generated IDs can repeat after restart or wraparound. ## Features Default features are `std`, `memory-queue`, and `nats-queue`; NATS implies `std`. The API executor requires `std`. -For `no_std` memory queues, set `default-features = false, features = ["memory-queue"]`. An allocator and pointer/32-bit atomics are required. This backend polls once, has no timeout support, and evicts old items when bounded; the `std` backend uses bounded Tokio channels and backpressure. Use positive capacities and unique task IDs: `TaskBuilder::default()` uses ID zero, and `no_std` generated IDs can repeat after restart or wraparound. +#### no_std -Custom queues/executors implement `QueueProducer`, `QueueWorker`, and `Executor`. Only the NATS/API pairing has a public `Worker` constructor; other combinations need a caller-managed loop. +For memory queues, use the feature `memory-queue`. An allocator and pointer/32-bit atomics are required. + +This backend polls once, has no timeout support, and evicts old items when bounded; the `std` backend uses bounded Tokio channels and backpressure. ## Development diff --git a/src/executor/openai/mod.rs b/src/executor/openai/mod.rs index 6c4e51f..248c35f 100644 --- a/src/executor/openai/mod.rs +++ b/src/executor/openai/mod.rs @@ -195,7 +195,6 @@ impl ExecutorAsyncOpenai { None } }) - .chain(iter::once(Interaction::User(prompt))) .chain(iter::once(Interaction::Assistant(m.clone()))) .collect(); diff --git a/src/queue/nats/mod.rs b/src/queue/nats/mod.rs index b044708..bb0a0f4 100644 --- a/src/queue/nats/mod.rs +++ b/src/queue/nats/mod.rs @@ -33,7 +33,7 @@ pub struct NatsProducer { } #[derive(Debug, Clone)] -/// Consumes one model subject in the fixed `task_workers` queue group. +/// Consumes one model subject in the workers queue group. /// /// Clones share one subscription. Replies require the incoming message's reply /// subject; malformed JSON and missing reply subjects return errors. @@ -103,8 +103,10 @@ impl NatsWorker { .connect(url) .await?; + let workers_group = + env::var("OPENAI_API_NATS_WORKERS_GROUP").unwrap_or_else(|_| "task_workers".into()); let subscriber = client - .queue_subscribe(subject.clone(), "task_workers".to_string()) + .queue_subscribe(subject.clone(), workers_group) .await?; let subscriber = Arc::new(Mutex::new(subscriber)); diff --git a/src/task.rs b/src/task.rs index 8b960f6..d678b79 100644 --- a/src/task.rs +++ b/src/task.rs @@ -7,7 +7,7 @@ use serde_json::Value; use crate::{queue::QueueProducer, utils}; -#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] /// Builds a chat task; only the prompt is required at build time. /// /// Prefer [`Self::new`] for a generated ID: derived `Default` uses ID zero. @@ -30,12 +30,24 @@ pub struct TaskBuilder { pub prompt: Option, } +impl Default for TaskBuilder { + fn default() -> Self { + Self::new() + } +} + impl TaskBuilder { /// Creates an empty builder with an ID from [`utils::id`]. pub fn new() -> Self { Self { id: utils::id(), - ..Default::default() + model: None, + max_tokens: None, + payload: None, + system: None, + history: None, + schema: None, + prompt: None, } }