#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtractedFact {
pub text: String,
pub entities: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtractedRelation {
pub subject: String,
pub predicate: String,
pub object: String,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ExtractedAttribute {
pub entity: String,
pub key: String,
pub value: serde_json::Value,
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Extraction {
pub facts: Vec<ExtractedFact>,
pub relations: Vec<ExtractedRelation>,
pub attributes: Vec<ExtractedAttribute>,
}
#[derive(Debug, thiserror::Error)]
pub enum ExtractError {
#[error("extraction backend error: {0}")]
Backend(String),
#[error("could not parse facts from extractor output: {0}")]
Parse(String),
}
pub trait Extractor {
fn extract(&self, text: &str) -> Result<Vec<ExtractedFact>, ExtractError>;
fn extract_graph(&self, text: &str) -> Result<Extraction, ExtractError> {
Ok(Extraction {
facts: self.extract(text)?,
..Extraction::default()
})
}
}
impl<T: Extractor + ?Sized> Extractor for std::sync::Arc<T> {
fn extract(&self, text: &str) -> Result<Vec<ExtractedFact>, ExtractError> {
(**self).extract(text)
}
fn extract_graph(&self, text: &str) -> Result<Extraction, ExtractError> {
(**self).extract_graph(text)
}
}
pub type DynExtractor = std::sync::Arc<dyn Extractor + Send + Sync>;
#[cfg(feature = "extract")]
pub const DEFAULT_OLLAMA_URL: &str = "http://localhost:11434";
#[cfg(feature = "extract")]
const REQUEST_TIMEOUT_SECS: u64 = 300;
#[cfg(feature = "extract")]
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
#[cfg(feature = "extract")]
const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[cfg(feature = "extract")]
const EXTRACT_LEVERS: crate::ollama_retry::OllamaLevers<'static> =
crate::ollama_retry::OllamaLevers {
url_var: "VELESDB_MEMORY_EXTRACTOR_URL",
model_var: "VELESDB_MEMORY_EXTRACTOR_MODEL",
fallback: None,
};
#[cfg(feature = "extract")]
enum GenerateCall {
Transport(Box<ureq::Error>),
Body(std::io::Error),
}
#[cfg(feature = "extract")]
fn generate_is_retryable(err: &GenerateCall) -> bool {
match err {
GenerateCall::Transport(inner) => crate::ollama_retry::is_retryable(inner),
GenerateCall::Body(inner) => crate::ollama_retry::io_is_retryable(inner),
}
}
#[cfg(feature = "extract")]
fn describe_generate_failure(url: &str, model: &str, err: &GenerateCall, attempts: u32) -> String {
let cause = match err {
GenerateCall::Transport(inner) => inner.to_string(),
GenerateCall::Body(inner) => format!("reading the response failed: {inner}"),
};
crate::ollama_retry::actionable_failure(
"generate",
url,
model,
attempts,
&cause,
&EXTRACT_LEVERS,
)
}
#[cfg(feature = "extract")]
#[derive(Debug, Clone)]
pub struct OllamaExtractor {
base_url: String,
model: String,
agent: ureq::Agent,
}
#[cfg(feature = "extract")]
impl OllamaExtractor {
#[must_use]
pub fn new(base_url: impl Into<String>, model: impl Into<String>) -> Self {
let timeout = std::time::Duration::from_secs(REQUEST_TIMEOUT_SECS);
let agent = ureq::AgentBuilder::new()
.timeout_connect(CONNECT_TIMEOUT)
.timeout_write(WRITE_TIMEOUT)
.timeout_read(timeout)
.timeout(timeout)
.build();
Self {
base_url: base_url.into(),
model: model.into(),
agent,
}
}
}
#[cfg(feature = "extract")]
impl Extractor for OllamaExtractor {
fn extract(&self, text: &str) -> Result<Vec<ExtractedFact>, ExtractError> {
let reply = self.generate(&build_prompt(text))?;
let raw = json_slice::<Vec<RawFact>>(&reply)
.ok_or_else(|| ExtractError::Parse(truncate(&reply)))?;
Ok(raw.into_iter().filter_map(RawFact::into_fact).collect())
}
fn extract_graph(&self, text: &str) -> Result<Extraction, ExtractError> {
let reply = self.generate(&build_graph_prompt(text))?;
let raw = json_slice_object::<RawExtraction>(&reply)
.ok_or_else(|| ExtractError::Parse(truncate(&reply)))?;
Ok(raw.into_extraction())
}
}
#[cfg(feature = "extract")]
impl OllamaExtractor {
fn generate(&self, prompt: &str) -> Result<String, ExtractError> {
let url = format!("{}/api/generate", self.base_url);
let body = serde_json::json!({
"model": self.model,
"prompt": prompt,
"stream": false,
"think": false,
"keep_alive": crate::embedder::keep_alive(),
"options": { "temperature": 0 },
})
.to_string();
let attempt = || {
let response = self
.agent
.post(&url)
.set("Content-Type", "application/json")
.send_string(&body)
.map_err(|err| GenerateCall::Transport(Box::new(err)))?;
response.into_string().map_err(GenerateCall::Body)
};
let payload = crate::ollama_retry::with_retry(
&crate::ollama_retry::OLLAMA_RETRIES,
generate_is_retryable,
attempt,
)
.map_err(|(err, attempts)| {
ExtractError::Backend(describe_generate_failure(&url, &self.model, &err, attempts))
})?;
parse_generate_response(&payload)
}
}
#[cfg(feature = "extract")]
#[derive(serde::Deserialize)]
struct RawFact {
fact: String,
#[serde(default)]
entities: Vec<String>,
}
#[cfg(feature = "extract")]
impl RawFact {
fn into_fact(self) -> Option<ExtractedFact> {
let text = self.fact.trim().to_string();
if text.is_empty() {
return None;
}
let mut entities: Vec<String> = self
.entities
.into_iter()
.map(|entity| entity.trim().to_lowercase())
.filter(|entity| !entity.is_empty())
.collect();
entities.sort_unstable();
entities.dedup();
Some(ExtractedFact { text, entities })
}
}
#[cfg(feature = "extract")]
fn canonical_entity(name: &str) -> String {
name.trim().to_lowercase()
}
#[cfg(feature = "extract")]
#[derive(serde::Deserialize)]
struct RawExtraction {
#[serde(default)]
facts: Vec<RawFact>,
#[serde(default)]
relations: Vec<RawRelation>,
#[serde(default)]
attributes: Vec<RawAttribute>,
}
#[cfg(feature = "extract")]
#[derive(serde::Deserialize)]
struct RawRelation {
subject: String,
predicate: String,
object: String,
}
#[cfg(feature = "extract")]
#[derive(serde::Deserialize)]
struct RawAttribute {
entity: String,
key: String,
value: serde_json::Value,
}
#[cfg(feature = "extract")]
impl RawExtraction {
fn into_extraction(self) -> Extraction {
Extraction {
facts: self
.facts
.into_iter()
.filter_map(RawFact::into_fact)
.collect(),
relations: self
.relations
.into_iter()
.filter_map(RawRelation::into_relation)
.collect(),
attributes: self
.attributes
.into_iter()
.filter_map(RawAttribute::into_attribute)
.collect(),
}
}
}
#[cfg(feature = "extract")]
impl RawRelation {
fn into_relation(self) -> Option<ExtractedRelation> {
let subject = canonical_entity(&self.subject);
let object = canonical_entity(&self.object);
let predicate = self.predicate.trim().to_string();
if subject.is_empty() || object.is_empty() || predicate.is_empty() || subject == object {
return None;
}
Some(ExtractedRelation {
subject,
predicate,
object,
})
}
}
#[cfg(feature = "extract")]
impl RawAttribute {
fn into_attribute(self) -> Option<ExtractedAttribute> {
let entity = canonical_entity(&self.entity);
let key = self.key.trim().to_string();
if entity.is_empty() || key.is_empty() || self.value.is_null() {
return None;
}
Some(ExtractedAttribute {
entity,
key,
value: self.value,
})
}
}
#[cfg(feature = "extract")]
fn build_graph_prompt(text: &str) -> String {
format!(
"You are building a knowledge graph from the passage below.\n\n\
Passage:\n{text}\n\n\
Return THREE things.\n\n\
1. \"facts\": the atomic, standalone facts a person would remember. Rewrite each \
as a self-contained sentence (resolve pronouns to names; keep absolute dates). \
For each, list 1-4 key TOPICS it concerns, as short canonical lowercase noun \
phrases, so the same topic recurs as the SAME tag across passages.\n\n\
2. \"relations\": every explicit relationship BETWEEN TWO NAMED ENTITIES, as \
subject/predicate/object triples. Use the entity's full name, lowercase \
(e.g. \"julien lange\").\n\
The predicate is a LABEL, not a sentence: **at most 3 words**, lowercase, in \
the passage's own language (e.g. \"pere de\", \"soeur de\", \"works at\", \
\"moteur de recherche\"). NEVER restate the sentence — write \"surveille les \
fuites\", not \"est utilise pour la surveillance de fuites de donnees\". If you \
cannot say it in 3 words, pick the closest short label.\n\
State the triple in the direction the passage states it, and add the converse \
ONLY if the passage states it too.\n\
Every named entity the passage RELATES to another must appear in at least one \
triple — an entity that only receives attributes and no edge is a dead end in \
the graph.\n\n\
3. \"attributes\": every property a named entity HAS, as entity/key/value. Use \
short lowercase keys (\"age\", \"ville\", \"employeur\"). Emit numbers as JSON \
NUMBERS, never strings: 15, not \"15\". Omit anything the passage does not state.\n\n\
Return ONLY this JSON object, no prose:\n\
{{\"facts\": [{{\"fact\": string, \"entities\": [string]}}], \
\"relations\": [{{\"subject\": string, \"predicate\": string, \"object\": string}}], \
\"attributes\": [{{\"entity\": string, \"key\": string, \"value\": string|number|boolean}}]}}"
)
}
#[cfg(feature = "extract")]
fn build_prompt(text: &str) -> String {
format!(
"You are building a memory graph from the passage below.\n\n\
Passage:\n{text}\n\n\
Extract the atomic, standalone facts a person would remember. Rewrite each as a \
self-contained sentence (resolve pronouns to names; keep absolute dates). For \
each fact also list 1-4 key TOPICS it concerns: the recurring subjects, \
activities, events, interests, plans, places, organisations, or named people a \
later question might reference. Use short, canonical, lowercase noun phrases \
(e.g. \"adoption\", \"charity race\", \"therapy\", \"new job\") so the same topic \
recurs as the SAME tag across passages.\n\n\
Return ONLY a JSON array, no prose, each item exactly:\n\
{{\"fact\": string, \"entities\": [string]}}"
)
}
#[cfg(feature = "extract")]
fn parse_generate_response(body: &str) -> Result<String, ExtractError> {
let value: serde_json::Value = serde_json::from_str(body)
.map_err(|err| ExtractError::Backend(format!("invalid generate response: {err}")))?;
let text = value
.get("response")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| ExtractError::Backend("ollama reply had no `response` field".to_string()))?;
Ok(text.trim().to_string())
}
#[cfg(feature = "extract")]
fn truncate(text: &str) -> String {
const LIMIT: usize = 120;
let mut out = String::new();
for word in text.split_whitespace() {
let sep_len = usize::from(!out.is_empty());
if out.len() + sep_len + word.len() > LIMIT {
break;
}
if !out.is_empty() {
out.push(' ');
}
out.push_str(word);
}
out
}
#[cfg(feature = "extract")]
fn json_slice<T: serde::de::DeserializeOwned>(text: &str) -> Option<T> {
let slice = balanced_slice(text)?;
serde_json::from_str::<T>(slice).ok()
}
#[cfg(feature = "extract")]
fn json_slice_object<T: serde::de::DeserializeOwned>(text: &str) -> Option<T> {
let slice = balanced_slice_preferring(text, b'{')?;
serde_json::from_str::<T>(slice).ok()
}
#[cfg(feature = "extract")]
fn balanced_slice(text: &str) -> Option<&str> {
balanced_slice_preferring(text, b'[')
}
#[cfg(feature = "extract")]
fn balanced_slice_preferring(text: &str, preferred: u8) -> Option<&str> {
let bytes = text.as_bytes();
let fallback = if preferred == b'[' { b'{' } else { b'[' };
let start = bytes
.iter()
.position(|&b| b == preferred)
.or_else(|| bytes.iter().position(|&b| b == fallback))?;
let open = bytes[start];
let close = if open == b'[' { b']' } else { b'}' };
let mut depth = 0u32;
let mut in_string = false;
let mut escaped = false;
for (offset, &byte) in bytes[start..].iter().enumerate() {
if in_string {
in_string = step_string(&mut escaped, byte);
} else if scan_structural(byte, open, close, &mut in_string, &mut depth) {
return Some(&text[start..=start + offset]);
}
}
None
}
#[cfg(feature = "extract")]
fn scan_structural(byte: u8, open: u8, close: u8, in_string: &mut bool, depth: &mut u32) -> bool {
if byte == b'"' {
*in_string = true;
} else if byte == open {
*depth += 1;
} else if byte == close {
*depth = depth.saturating_sub(1);
return *depth == 0;
}
false
}
#[cfg(feature = "extract")]
fn step_string(escaped: &mut bool, byte: u8) -> bool {
match (*escaped, byte) {
(true, _) => {
*escaped = false;
true
}
(false, b'\\') => {
*escaped = true;
true
}
(false, b'"') => false,
(false, _) => true,
}
}
#[cfg(all(test, feature = "extract"))]
mod tests {
use super::*;
#[test]
fn parses_a_graph_reply_whose_first_bracket_is_nested() {
let reply = r#"{ "facts": [ { "fact": "Zephyrin is the father of Kaltar.", "entities": ["zephyrin", "kaltar"] } ], "relations": [ { "subject": "zephyrin", "predicate": "pere de", "object": "kaltar" } ], "attributes": [ { "entity": "kaltar", "key": "age", "value": 15 } ] }"#;
let raw: RawExtraction = json_slice_object(reply).expect("the object is sliced whole");
let extraction = raw.into_extraction();
assert_eq!(extraction.facts.len(), 1);
assert_eq!(extraction.relations.len(), 1);
assert_eq!(extraction.relations[0].predicate, "pere de");
assert_eq!(extraction.attributes.len(), 1);
assert_eq!(extraction.attributes[0].value, serde_json::json!(15));
}
#[test]
fn parses_a_graph_reply_wrapped_in_prose_and_fences() {
let reply = "Here you go:\n```json\n{\"facts\": [], \"relations\": [{\"subject\": \"a\", \"predicate\": \"knows\", \"object\": \"b\"}], \"attributes\": []}\n```";
let raw: RawExtraction = json_slice_object(reply).expect("sliced past the fence");
assert_eq!(raw.into_extraction().relations.len(), 1);
}
#[test]
fn fact_only_slicing_still_prefers_the_array() {
let reply = "Result {ok}: [{\"fact\": \"A ships B.\", \"entities\": [\"b\"]}]";
let raw: Vec<RawFact> = json_slice(reply).expect("array sliced despite the stray brace");
assert_eq!(raw.len(), 1);
}
#[test]
fn graph_prompt_demands_numeric_values_and_the_three_sections() {
let prompt = build_graph_prompt("Kaltar a 15 ans.");
assert!(prompt.contains("Kaltar a 15 ans."));
assert!(prompt.contains("\"relations\""));
assert!(prompt.contains("\"attributes\""));
assert!(prompt.contains("15, not \"15\""));
}
#[test]
fn prompt_carries_the_passage_and_json_contract() {
let prompt = build_prompt("Alice adopted a dog in 2021.");
assert!(prompt.contains("Alice adopted a dog in 2021."));
assert!(prompt.contains("\"fact\": string"));
}
#[test]
fn graph_prompt_bounds_the_predicate_and_demands_edges() {
let prompt = build_graph_prompt("Ahmia is an onion search engine.");
assert!(prompt.contains("Ahmia is an onion search engine."));
assert!(
prompt.contains("at most 3 words"),
"the predicate length must be a hard bound, not a suggestion"
);
assert!(
prompt.contains("NEVER restate the sentence"),
"the counter-example is what stops a restated sentence"
);
assert!(
prompt.contains("at least one triple"),
"an entity with attributes but no edge is a dead end — the prompt \
must ask for the edge"
);
}
#[test]
fn parses_facts_from_a_fenced_reply() {
let reply = "Sure!\n```json\n[{\"fact\":\"Alice adopted a dog.\",\"entities\":[\"Adoption\",\"adoption\",\"\"]}]\n```";
let facts: Vec<RawFact> = json_slice(reply).expect("slice json");
let facts: Vec<ExtractedFact> = facts.into_iter().filter_map(RawFact::into_fact).collect();
assert_eq!(facts.len(), 1);
assert_eq!(facts[0].text, "Alice adopted a dog.");
assert_eq!(facts[0].entities, vec!["adoption".to_string()]);
}
#[test]
fn drops_a_textless_fact() {
let raw = RawFact {
fact: " ".to_string(),
entities: vec!["x".to_string()],
};
assert!(raw.into_fact().is_none());
}
#[test]
fn parses_response_envelope() {
let text = parse_generate_response(r#"{"response":" [] "}"#).expect("parse");
assert_eq!(text, "[]");
}
#[test]
fn rejects_response_without_field() {
assert!(matches!(
parse_generate_response(r#"{"oops":true}"#),
Err(ExtractError::Backend(_))
));
}
}