5 sections·~3 min read·1 expanded
Instead of storing "User: We're about 50 people right now, growing fast though" verbatim, you want an LLM to extract: {"fact": "Company has approximately 50 employees", "confidence": "stated by user", "type": "semantic"}.
import json
from openai import OpenAI
client = OpenAI()
EXTRACTION_PROMPT = """Extract discrete factual statements from this conversation turn.
Return a JSON list of objects, each with "fact" (a clear standalone statement)
and "type" (one of: episodic, semantic, procedural).
Only extract genuinely new information, not questions or filler.
Conversation turn:
{turn}
JSON:"""
def extract_facts(turn):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": EXTRACTION_PROMPT.format(turn=turn)}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
facts = extract_facts("We're about 50 people right now, growing fast though.")
print(facts)
For every new candidate fact, before blindly storing it, check it against similar existing memories and decide what to do:
ROUTING_PROMPT = """You are managing a memory store. Given a NEW fact and a list
of EXISTING similar facts already stored, decide the correct operation:
- ADD: the new fact is genuinely new information
- UPDATE: the new fact refines or corrects an existing one (specify which existing fact id)
- DELETE: the new fact means an existing fact is no longer true (specify which existing fact id)
- NOOP: the new fact adds nothing new
NEW fact: {new_fact}
EXISTING facts: {existing_facts}
Respond with JSON: {{"operation": "...", "target_id": "..." or null, "reasoning": "..."}}"""
def route_fact(new_fact, existing_facts):
prompt = ROUTING_PROMPT.format(new_fact=new_fact, existing_facts=existing_facts)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
# Example: this should trigger an UPDATE
existing = [{"id": "fact_1", "text": "Company has approximately 50 employees"}]
new = "We just closed a round, we're closer to 80 employees now"
print(route_fact(new, existing))
# Expect: {"operation": "UPDATE", "target_id": "fact_1", "reasoning": "..."}
import chromadb
chroma_client = chromadb.Client()
memory_collection = chroma_client.create_collection(name="agent_memory")
def get_similar_facts(new_fact_text, k=3):
if memory_collection.count() == 0:
return []
results = memory_collection.query(query_texts=[new_fact_text], n_results=k)
return [{"id": id_, "text": doc} for id_, doc in
zip(results['ids'][0], results['documents'][0])]
def process_new_fact(fact_text):
similar = get_similar_facts(fact_text)
decision = route_fact(fact_text, similar)
if decision["operation"] == "ADD":
new_id = f"fact_{memory_collection.count()}"
memory_collection.add(documents=[fact_text], ids=[new_id])
print(f"ADDED: {fact_text}")
elif decision["operation"] == "UPDATE":
memory_collection.update(ids=[decision["target_id"]], documents=[fact_text])
print(f"UPDATED {decision['target_id']}: {fact_text}")
elif decision["operation"] == "DELETE":
memory_collection.delete(ids=[decision["target_id"]])
print(f"DELETED {decision['target_id']}")
else:
print(f"NOOP: {fact_text}")
process_new_fact("Company has approximately 50 employees")
process_new_fact("We just closed a round, we're closer to 80 employees now")
Run this — you should see the first call ADD, and the second call UPDATE (overwriting the same memory slot rather than creating a duplicate, contradicting entry).
Every single new fact requires an LLM call to classify it. If your agent processes thousands of conversations, this routing cost adds up fast — this is exactly why "novelty gating" (skip the LLM call for obviously-new, obviously-irrelevant facts using a cheap similarity threshold check first) becomes important at scale (covered more in Module 10).
- Build the full pipeline above yourself, don't just read it — run it on a 10-turn synthetic conversation where the user states, then later contradicts, 3 different facts.
- Verify all 3 contradictions correctly trigger UPDATE, not duplicate ADD.
- Deliberately feed it a fact that should be a NOOP (something already effectively stored) and verify it's correctly ignored.