feat: POST /memory/learn endpoint + refactor mem learn CLI

Learning flow now goes through the service, not local JSONL:
- POST /memory/learn: accepts markdown, chunks it, runs gated loop
  (LLM evaluates + compacts), stores in pgvector. OpenAI-style API.
- mem learn CLI: reads files, calls POST /memory/learn per file
- Removed cmd_compact (gated loop IS the compaction)
- Updated README with new commands and API docs

Memory never grows unbounded — every update is a rewrite, not append.
The gated loop LLM acts as evaluator + compactor in one pass.
This commit is contained in:
2026-08-30 13:21:07 -07:00
parent 92458e643c
commit ae1a2ef9a2
3 changed files with 328 additions and 287 deletions
+63 -8
View File
@@ -4,8 +4,8 @@ Gated recurrent memory over agent context. Reads session history chunk-by-chunk,
keeps only what answers standing questions, projects result into an Obsidian
vault and a pgvector index.
**Status: design complete, no code yet.** 37 tasks in [memory-tasks/](memory-tasks/INDEX.md),
0 done. Start at [M0.1](memory-tasks/M0.1-cargo-workspace.md).
**Status: 78/78 tasks complete, all 13 phases done.** Production-deployed on Kubernetes
via ArgoCD. See [CLAUDE.md](CLAUDE.md) for full API reference.
## Problem
@@ -171,16 +171,71 @@ with an EOF; telemetry or a live tail will not have one. `RecordSource` returns
## Commands
```sh
mem ingest --project poimen --dry-run # chunk plan, zero model calls
# Ingest knowledge via gated loop (LLM evaluates + compacts automatically)
mem learn knowledge/rust.md # single file
mem learn knowledge/ --project myproject # directory
mem learn knowledge/ --dry-run # preview chunks
mem learn knowledge/ --memory-budget 8192 # larger memory window
mem learn knowledge/ --model ornith:35b # use stronger model
# Traditional ingest (from session transcripts)
mem ingest --project poimen --dry-run
mem ingest --project poimen --query infra-root-causes
mem synthesize --project poimen # L2 pass, exit gate on
mem rebuild --from-log --project poimen # drop and rebuild projections
mem verify --project poimen # provenance graph closure
mem query "why did requests over 10KB fail?"
# Failure capture + lesson derivation
mem capture --cmd "cargo build" --exit 1 --output-file error.log
mem sig --tool cargo --file error.log # extract failure signature
mem resolve --json # pair failure with fix
mem lookup --tool cargo --file error.log # search known fixes
# Skills + projections
mem skill draft --from poimen/infra-root-causes
mem label --project poimen # evidence labels for training
mem materialize # generate SKILL.md files
mem verify --project poimen # provenance graph closure
# Server
mem serve --port 8080
```
## HTTP API
```sh
# Health
curl http://localhost:8080/health
# Learn — gated loop ingest (LLM evaluates + compacts)
curl -X POST http://localhost:8080/memory/learn \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"project": "knowledge", "text": "## Rust\n- ownership...", "query": "key patterns?"}'
# Returns: {chunks_seen, chunks_used, memory: "compacted...", stored: true}
# Ingest — queue-based async ingest
curl -X POST http://localhost:8080/memory/ingest ...
# Query — hybrid semantic + lexical search
curl http://localhost:8080/memory/query?project=poimen&q=port+conflict
# Context — three-tier retrieval (signature > vector > reference)
curl -X POST http://localhost:8080/memory/context \
-d '{"project": "poimen", "tool": "cargo", "task": "build", "budget": 4096}'
```
### Learning Flow
```
Agent/CLI → POST /memory/learn → chunk markdown → gated loop:
For each chunk:
LLM evaluates: does this add new knowledge? (update gate)
If yes → LLM rewrites memory incorporating new fact (compaction)
If no → chunk rejected, memory unchanged
→ Store compacted memory in pgvector (embedded, searchable)
→ Return {chunks_seen, chunks_used, memory, stored}
```
Memory never grows unbounded — every update is a rewrite, not an append.
The LLM acts as both evaluator and compactor in one pass.
## M3.8 Pluggable Query Optimization
**Purpose**: Compress and optimize search results before passing them to the LLM context window, improving token efficiency and response quality.
+188
View File
@@ -357,6 +357,7 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res
.route("/memory/context", web::post().to(context_handler))
.route("/memory/projects", web::get().to(projects_handler))
.route("/memory/skills", web::get().to(skills_handler))
.route("/memory/learn", web::post().to(learn_handler))
.route("/memory/vault/generate", web::post().to(vault_generate_handler))
.route("/memory/vault", web::get().to(vault_browser_handler))
.route("/memory/vault/{project}", web::get().to(vault_project_handler))
@@ -562,6 +563,193 @@ async fn optimize_search_results(
optimized
}
/// POST /memory/learn — Ingest knowledge via gated loop (LLM evaluates + compacts)
///
/// OpenAI-compatible style endpoint. Accepts markdown text, chunks it,
/// runs each chunk through the gated loop where the LLM decides whether
/// to accept/reject and rewrites memory to stay compact.
///
/// Request:
/// POST /memory/learn
/// { "project": "knowledge", "text": "## Rust\n- ownership...", "query": "What are key Rust patterns?" }
///
/// Response:
/// { "project": "knowledge", "chunks_seen": 5, "chunks_used": 3,
/// "memory": "compacted memory text...", "status": "completed" }
pub async fn learn_handler(
req: HttpRequest,
body: web::Json<serde_json::Value>,
state: web::Data<AppState>,
) -> HttpResponse {
let (claims, _token) = match validate_auth(&req, &state).await {
Ok(c) => c,
Err(e) => return e,
};
if !has_capability(&claims, "memory:write") {
return HttpResponse::Forbidden().json(json!({
"error": "forbidden",
"reason": "missing capability: memory:write"
}));
}
if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
return e;
}
let project = body["project"].as_str().unwrap_or("knowledge").to_string();
let text = match body["text"].as_str() {
Some(t) => t.to_string(),
None => return HttpResponse::BadRequest().json(json!({
"error": "bad_request",
"reason": "missing required field: text"
})),
};
let question = body["query"].as_str()
.unwrap_or("What are the key facts, patterns, and practices in this knowledge?")
.to_string();
let memory_budget = body["memory_budget"].as_u64().unwrap_or(4096) as u32;
let chunk_size = body["chunk_size"].as_u64().unwrap_or(2000) as usize;
let model = body["model"].as_str().unwrap_or("qwen2.5:3b-instruct").to_string();
// Chunk the markdown
let chunks = chunk_markdown_text(&text, chunk_size);
if chunks.is_empty() {
return HttpResponse::BadRequest().json(json!({
"error": "bad_request",
"reason": "text produced no chunks"
}));
}
// Build domain chunks
let domain_chunks: Vec<mem_core::domain::Chunk> = chunks
.iter()
.enumerate()
.map(|(i, text)| {
let record = mem_core::Record {
role: mem_core::Role::User,
text: text.clone(),
timestamp: time::OffsetDateTime::now_utc(),
provenance: mem_core::Provenance {
source_id: format!("learn://{}:{}", project, i),
offset: i as u64,
},
};
mem_core::domain::Chunk::new((i + 1) as u32, vec![record], text.len() / 4)
})
.collect();
// Build query for gated loop
let query = mem_core::query::Query {
id: format!("learn-{}", uuid::Uuid::new_v4()),
question,
exit_gate: false,
};
let config = mem_core::gated_loop::LoopConfig {
level: mem_core::Level::L1,
query,
memory_budget,
use_exit_gate: false,
};
// Create LLM client
let llm_base = std::env::var("LLM_API_BASE")
.unwrap_or_else(|_| "https://api.riotpiao.com".to_string());
let llm_key = std::env::var("LLM_API_KEY")
.or_else(|_| std::env::var("MEM_API_KEY"))
.unwrap_or_default();
let llm = match mem_llm::ChatClient::new(&llm_base, &llm_key, &model) {
Ok(c) => c,
Err(e) => return HttpResponse::InternalServerError().json(json!({
"error": "llm_init_failed",
"reason": e.to_string()
})),
};
// Run gated loop
let outcome = match mem_core::gated_loop::run_loop(config, domain_chunks, &llm) {
Ok(o) => o,
Err(e) => return HttpResponse::InternalServerError().json(json!({
"error": "gated_loop_failed",
"reason": e.to_string()
})),
};
// Store compacted memory in pgvector if non-empty
let mut stored = false;
if !outcome.final_memory.is_empty() {
match state.embeddings.embed_one(&outcome.final_memory).await {
Ok(embedding) => {
let chunk_id = uuid::Uuid::new_v4();
let sha = {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(outcome.final_memory.as_bytes());
format!("{:x}", h.finalize())
};
let result = sqlx::query(
"INSERT INTO memory_chunks (id, project, level, text, embedding, sha256, source, created_at)
VALUES ($1, $2, 'L1', $3, $4, $5, $6, NOW())
ON CONFLICT (sha256) DO UPDATE SET text = $3, embedding = $4",
)
.bind(chunk_id)
.bind(&project)
.bind(&outcome.final_memory)
.bind(embedding.to_vec())
.bind(&sha)
.bind(format!("learn://{}", project))
.fetch_optional(&state.pool)
.await;
match result {
Ok(_) => { stored = true; }
Err(e) => {
tracing::error!("Failed to store compacted memory: {}", e);
}
}
}
Err(e) => {
tracing::error!("Failed to embed compacted memory: {}", e);
}
}
}
HttpResponse::Ok().json(json!({
"project": project,
"status": "completed",
"chunks_seen": outcome.chunks_seen,
"chunks_used": outcome.chunks_used,
"memory": outcome.final_memory,
"memory_tokens": outcome.final_memory.len() / 4,
"stored": stored,
"model": model,
}))
}
/// Split markdown on ## headings for learn endpoint.
fn chunk_markdown_text(content: &str, max_chunk: usize) -> Vec<String> {
let mut chunks = Vec::new();
let mut current = String::new();
for line in content.lines() {
if line.starts_with("## ") && !current.is_empty() {
let trimmed = current.trim().to_string();
if !trimmed.is_empty() { chunks.push(trimmed); }
current = String::new();
}
current.push_str(line);
current.push('\n');
if current.len() > max_chunk {
let trimmed = current.trim().to_string();
if !trimmed.is_empty() { chunks.push(trimmed); }
current = String::new();
}
}
let trimmed = current.trim().to_string();
if !trimmed.is_empty() { chunks.push(trimmed); }
chunks
}
/// GET /memory/query — semantic search across memories
pub async fn query_handler(
req: HttpRequest,
+77 -279
View File
@@ -156,22 +156,7 @@ enum Commands {
file: Option<PathBuf>,
},
/// Compact knowledge: deduplicate and merge similar chunks via embeddings + LLM
Compact {
/// Project name
#[arg(long, default_value = "knowledge")]
project: String,
/// Cosine similarity threshold for grouping
#[arg(long, default_value_t = 0.82)]
threshold: f32,
/// Dry run — show groups without merging
#[arg(long)]
dry_run: bool,
},
/// Ingest markdown knowledge files into memory
/// Ingest markdown knowledge via gated loop (LLM evaluates + compacts)
Learn {
/// Markdown files or directories to ingest
#[arg(value_name = "PATH")]
@@ -181,13 +166,21 @@ enum Commands {
#[arg(long, default_value = "knowledge")]
project: String,
/// Dry run — show chunks without writing
/// Dry run — show chunks without ingesting
#[arg(long)]
dry_run: bool,
/// Maximum chunk size in characters (splits on headings)
#[arg(long, default_value_t = 2000)]
chunk_size: usize,
/// Memory budget in tokens (LLM compacts to fit)
#[arg(long, default_value_t = 4096)]
memory_budget: u32,
/// LLM model for gated evaluation
#[arg(long, default_value = "qwen2.5:3b-instruct")]
model: String,
},
}
@@ -247,11 +240,8 @@ async fn main() -> anyhow::Result<()> {
Commands::Sig { tool, file } => {
cmd_sig(&tool, file.as_ref())?
}
Commands::Learn { paths, project, dry_run, chunk_size } => {
cmd_learn(&paths, &project, dry_run, chunk_size)?;
}
Commands::Compact { project, threshold, dry_run } => {
cmd_compact(&project, threshold, dry_run).await?;
Commands::Learn { paths, project, dry_run, chunk_size, memory_budget, model } => {
cmd_learn(&paths, &project, dry_run, chunk_size, memory_budget, &model).await?;
}
}
@@ -419,215 +409,15 @@ async fn cmd_verify(
Ok(())
}
async fn cmd_compact(project: &str, threshold: f32, dry_run: bool) -> anyhow::Result<()> {
use mem_llm::{EmbeddingsClient, ChatClient};
use sha2::{Digest, Sha256};
let log_path = format!("log/{}/learn/latest.jsonl", project);
let content = fs::read_to_string(&log_path)
.map_err(|_| anyhow::anyhow!("No log at {}", log_path))?;
let mut records: Vec<serde_json::Value> = content
.lines()
.filter(|l| !l.is_empty())
.map(|l| serde_json::from_str(l).unwrap())
.collect();
let texts: Vec<String> = records
.iter()
.map(|r| r["data"]["text"].as_str().unwrap_or("").to_string())
.collect();
println!("Loaded {} chunks from {}", records.len(), log_path);
// Embed all chunks
println!("Embedding {} chunks...", texts.len());
let embedder = EmbeddingsClient::from_env()?;
let mut vectors = Vec::new();
for batch in texts.chunks(8) {
let batch_strs: Vec<String> = batch.to_vec();
match embedder.embed(&batch_strs).await {
Ok(v) => vectors.extend(v),
Err(e) => {
eprintln!("Embedding batch failed: {}. Retrying in 5s...", e);
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
let v = embedder.embed(&batch_strs).await?;
vectors.extend(v);
}
}
eprint!(".");
}
eprintln!();
println!("Embedded {} vectors (768-dim)", vectors.len());
// Compute cosine similarity and group
let n = vectors.len();
let raw_vecs: Vec<Vec<f32>> = vectors.iter().map(|v| v.to_vec()).collect();
// Normalize vectors
let norms: Vec<f32> = raw_vecs
.iter()
.map(|v| {
let s: f32 = v.iter().map(|x| x * x).sum();
s.sqrt().max(1e-10)
})
.collect();
let normed: Vec<Vec<f32>> = raw_vecs
.iter()
.zip(norms.iter())
.map(|(v, n)| v.iter().map(|x| x / n).collect())
.collect();
// Find similar groups
let mut visited = vec![false; n];
let mut groups: Vec<Vec<usize>> = Vec::new();
for i in 0..n {
if visited[i] { continue; }
let mut group = vec![i];
visited[i] = true;
for j in (i + 1)..n {
if visited[j] { continue; }
let sim: f32 = normed[i].iter().zip(normed[j].iter()).map(|(a, b)| a * b).sum();
if sim > threshold {
group.push(j);
visited[j] = true;
}
}
if group.len() > 1 {
groups.push(group);
}
}
if groups.is_empty() {
println!("\n\u{2713} No similar chunks found. Knowledge is already compact.");
return Ok(());
}
let total_mergeable: usize = groups.iter().map(|g| g.len()).sum();
let savings = total_mergeable - groups.len();
println!("\nFound {} groups ({} chunks \u{2192} {} merged, saving {})",
groups.len(), total_mergeable, groups.len(), savings);
let llm = ChatClient::new(
std::env::var("LLM_API_BASE").unwrap_or_else(|_| "https://api.riotpiao.com".to_string()),
std::env::var("LLM_API_KEY").unwrap_or_default(),
"reasoning",
)?;
let mut to_remove: Vec<usize> = Vec::new();
for (gi, group) in groups.iter().enumerate() {
let group_texts: Vec<&str> = group.iter().map(|&i| texts[i].as_str()).collect();
let group_sources: Vec<&str> = group
.iter()
.map(|&i| records[i]["data"]["source"].as_str().unwrap_or("?"))
.collect();
// Compute max similarity in group
let mut max_sim: f32 = 0.0;
for a in 0..group.len() {
for b in (a + 1)..group.len() {
let sim: f32 = normed[group[a]].iter().zip(normed[group[b]].iter()).map(|(x, y)| x * y).sum();
max_sim = max_sim.max(sim);
}
}
println!("\nGroup {} (sim={:.3}, {} chunks):", gi + 1, max_sim, group.len());
for &idx in group {
let preview: String = texts[idx].chars().take(80).collect();
let src = std::path::Path::new(group_sources[group.iter().position(|&i| i == idx).unwrap()])
.file_stem()
.map(|s| s.to_string_lossy().to_string())
.unwrap_or_else(|| "?".to_string());
println!(" [{}] {}...", src, preview.replace('\n', " "));
}
if dry_run {
continue;
}
// Merge via LLM
println!(" \u{2192} Merging with reasoning model...");
let numbered: String = group_texts
.iter()
.zip(group_sources.iter())
.enumerate()
.map(|(i, (t, s))| format!("[Chunk {} from {}]:\n{}", i + 1, s, t))
.collect::<Vec<_>>()
.join("\n\n");
let merged = llm.complete(
"Merge these similar knowledge chunks into ONE concise chunk. Keep ALL unique facts. \
Remove redundancy. Keep markdown formatting. Output ONLY the merged text.",
&format!("Merge these {} chunks:\n\n{}", group.len(), numbered),
1500,
).await?;
// Strip <think> tags
let merged_text = merged.text.split("</think>").last().unwrap_or(&merged.text).trim().to_string();
let old_size: usize = group_texts.iter().map(|t| t.len()).sum();
println!(" \u{2192} Merged: {} chars (was {} chars, {:.0}% reduction)",
merged_text.len(), old_size, (1.0 - merged_text.len() as f64 / old_size as f64) * 100.0);
// Update first chunk with merged content
let mut hasher = Sha256::new();
hasher.update(merged_text.as_bytes());
let new_hash = format!("{:x}", hasher.finalize());
records[group[0]]["data"]["text"] = serde_json::Value::String(merged_text);
records[group[0]]["data"]["sha256"] = serde_json::Value::String(new_hash);
records[group[0]]["data"]["merged_from"] = serde_json::json!(group.len());
// Mark rest for removal
for &idx in &group[1..] {
to_remove.push(idx);
}
}
if dry_run {
println!("\n(dry run \u{2014} no changes written)");
return Ok(());
}
// Write compacted log
let to_remove_set: std::collections::HashSet<usize> = to_remove.into_iter().collect();
let compacted: Vec<&serde_json::Value> = records
.iter()
.enumerate()
.filter(|(i, _)| !to_remove_set.contains(i))
.map(|(_, r)| r)
.collect();
// Backup
let backup = format!("{}.bak", log_path);
fs::copy(&log_path, &backup)?;
// Write
let mut f = fs::File::create(&log_path)?;
use std::io::Write;
for r in &compacted {
serde_json::to_writer(&mut f, r)?;
f.write_all(b"\n")?;
}
println!("\n{}", "\u{2500}".repeat(50));
println!("Before: {} chunks", records.len());
println!("After: {} chunks (-{})", compacted.len(), records.len() - compacted.len());
println!("Backup: {}", backup);
println!("Written: {}", log_path);
Ok(())
}
fn cmd_learn(
async fn cmd_learn(
paths: &[PathBuf],
project: &str,
dry_run: bool,
max_chunk: usize,
memory_budget: u32,
model: &str,
) -> anyhow::Result<()> {
use sha2::{Digest, Sha256};
let mut all_files: Vec<PathBuf> = Vec::new();
for p in paths {
if p.is_dir() {
@@ -658,73 +448,81 @@ fn cmd_learn(
all_files.sort();
println!("Found {} markdown files", all_files.len());
let mut total_chunks = 0usize;
let mut total_bytes = 0usize;
let mut log = if !dry_run {
Some(mem_store::LogWriter::new(project, "learn", "latest")?)
} else {
None
};
let api_url = std::env::var("MEM_API_URL")
.unwrap_or_else(|_| "http://localhost:8080".to_string());
let api_token = std::env::var("MEM_API_TOKEN").ok();
let http = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(120))
.build()?;
let mut total_seen = 0u64;
let mut total_used = 0u64;
for file in &all_files {
let content = fs::read_to_string(file)?;
let filename = file.file_stem().unwrap().to_string_lossy();
let chunks = chunk_markdown(&content, max_chunk);
println!("\n\u{1f4c4} {} \u{2014} {} chunks", file.display(), chunks.len());
println!("\n📄 {}{} chunks", file.display(), chunks.len());
for (i, chunk) in chunks.iter().enumerate() {
let mut hasher = Sha256::new();
hasher.update(chunk.as_bytes());
let hash = format!("{:x}", hasher.finalize());
let short_hash = &hash[..12];
total_chunks += 1;
total_bytes += chunk.len();
if dry_run {
if dry_run {
for (i, chunk) in chunks.iter().enumerate() {
let preview: String = chunk.chars().take(80).collect();
println!(
" [{}/{}] {} ({} bytes) {}",
i + 1,
chunks.len(),
short_hash,
chunk.len(),
preview.replace('\n', " ")
);
} else {
let record = mem_store::EventRecord {
project: project.to_string(),
query: format!("{}:{}", filename, i),
run: "latest".to_string(),
turn: i as u32,
event_type: "learn".to_string(),
data: serde_json::json!({
"source": file.to_string_lossy(),
"chunk_index": i,
"total_chunks": chunks.len(),
"sha256": hash,
"level": "L1",
"text": chunk,
}),
};
log.as_mut().unwrap().log(record)?;
println!(" ✓ [{}/{}] {} ({} bytes)", i + 1, chunks.len(), short_hash, chunk.len());
println!(" [{}/{}] ({} bytes) {}",
i + 1, chunks.len(), chunk.len(),
preview.replace('\n', " "));
}
continue;
}
let payload = serde_json::json!({
"project": project,
"text": content,
"query": format!("What are the key facts and patterns in {}?", filename),
"memory_budget": memory_budget,
"chunk_size": max_chunk,
"model": model,
});
let mut req = http.post(format!("{}/memory/learn", api_url))
.header("Content-Type", "application/json")
.json(&payload);
if let Some(ref token) = api_token {
req = req.header("Authorization", format!("Bearer {}", token));
}
println!(" \u{2192} Sending to {}/memory/learn (model={})...", api_url, model);
match req.send().await {
Ok(resp) => {
let status = resp.status();
let body: serde_json::Value = resp.json().await.unwrap_or_default();
if status.is_success() {
let seen = body["chunks_seen"].as_u64().unwrap_or(0);
let used = body["chunks_used"].as_u64().unwrap_or(0);
let stored = body["stored"].as_bool().unwrap_or(false);
let mem_preview: String = body["memory"].as_str()
.unwrap_or("").chars().take(100).collect();
total_seen += seen;
total_used += used;
println!(" \u{2713} {} seen, {} accepted, stored={}", seen, used, stored);
println!(" \u{2713} Memory: {}...", mem_preview.replace('\n', " "));
} else {
eprintln!(" \u{2717} {} \u{2014} {}", status, body);
}
}
Err(e) => {
eprintln!(" \u{2717} API call failed: {}", e);
eprintln!(" Is the memory service running at {}?", api_url);
}
}
}
println!("\n{}", "".repeat(50));
println!(
"{} files → {} chunks ({:.1} KB)",
all_files.len(),
total_chunks,
total_bytes as f64 / 1024.0
);
println!("\n{}", "\u{2500}".repeat(50));
if dry_run {
println!("(dry run nothing written)");
println!("(dry run \u{2014} nothing sent)");
} else {
println!("Written to log/{}/learn/latest.jsonl", project);
println!("{} files \u{2192} {} chunks seen, {} accepted",
all_files.len(), total_seen, total_used);
println!("Ingested via gated loop at {}", api_url);
}
Ok(())
}