Cookbook¶
From the source repository
Every recipe's full code and real output are on this page: copy one into a file and run it after pip install kioko. File names such as examples/cookbook/01_profile.py refer to the source repository, which is not public.
Runnable recipes for every kind of memory, the intelligence, the stores and the service. Every recipe runs
offline (python examples/cookbook/<file>.py from the repository root), and the output shown under each one is
real: this page is generated from the files by python docs/build_cookbook.py (ids, ports and timings are
replaced by placeholders). tests/test_cookbook.py runs every recipe on every test run.
New to Kioko? Read the guide first; memory types explains each kind and intelligence the self-managing mode.
The memory kinds
- 01 · Profile memory
- 02 · Factual memory
- 03 · Semantic memory
- 04 · Episodic memory
- 05 · Working memory
- 06 · Conversational memory
- 07 · Goal memory
- 08 · Procedural memory
- 09 · Skill memory
- 10 · Strategy memory
- 11 · Decision memory
- 12 · Graph memory
- 13 · KV-cache memory
- 14 · Parametric memory
- 15 · Hybrid memory
Keeping memory healthy
Stores and the service
Putting it together
- 19 · A complete agent turn with Kioko
- 20 · Self-managing memory in one line
- 21 · Model-driven memory management
- 22 · How much the intelligence saves
- 23 · Memory graph
The memory kinds¶
01 Profile memory¶
Who the user is - one value per slot, older values kept as history.
examples/cookbook/01_profile.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.profile.set("name", "Kumar Nishant", user_id=user)
memory.profile.set("role", "senior AI engineer", user_id=user)
memory.profile.set("employer", "Rockwell Automation", user_id=user)
memory.profile.set("location", "Pune", user_id=user)
memory.profile.set("location", "Bengaluru", user_id=user) # moved: the slot is updated, not duplicated
print(memory.profile.get(user_id=user))
print("location history:", [h["content"] for h in memory.profile.history("location", user_id=user)])
print(memory.profile.render(user_id=user))
# Conversations fill the same slots automatically
memory.learn("By the way, I'm now based in Hyderabad.", user_id=user)
print("after learn():", memory.profile.get("location", user_id=user))
Output:
{'employer': 'Rockwell Automation', 'location': 'Bengaluru', 'name': 'Kumar Nishant', 'role': 'senior AI engineer'}
location history: ['User lives in Pune.']
- User works at Rockwell Automation.
- User lives in Bengaluru.
- User's name is Kumar Nishant.
- User is a senior AI engineer.
after learn(): Hyderabad
02 Factual memory¶
Facts and preferences, with confidence and verification.
examples/cookbook/02_facts.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.facts.add("The Pune plant runs ControlLogix 5580 controllers", user_id=user, confidence=0.7,
topics=["plant"])
memory.facts.add("User prefers answers with code first", user_id=user, preference=True)
manager = memory.facts.add("User's manager is Krutika", user_id=user, key="work.manager")
memory.facts.verify(manager, user_id=user, confidence=0.99) # confirmed by the user / a system of record
for record in memory.facts.list(user_id=user):
print(f"[{record.memory_type}] {record.content} {record.metadata}")
hits = memory.facts.search("which controllers does the plant use?", user_id=user)
print("best match:", hits[0].record.content, round(hits[0].score, 2))
# Contradictions supersede instead of piling up
memory.add("User likes long explanations", user_id=user, memory_type="preference")
memory.add("User doesn't like long explanations", user_id=user, memory_type="preference")
print([r.content for r in memory.facts.list(user_id=user) if "explanations" in r.content])
Output:
[fact] User's manager is Krutika {'verified': True, 'confidence': 0.99}
[preference] User prefers answers with code first {}
[fact] The Pune plant runs ControlLogix 5580 controllers {'confidence': 0.7}
best match: The Pune plant runs ControlLogix 5580 controllers 0.54
["User doesn't like long explanations"]
03 Semantic memory¶
Domain knowledge - shared by every user, or personal.
examples/cookbook/03_semantic.py
from kioko import MemoryLayer
memory = MemoryLayer()
# A shared knowledge base (stored under the reserved user "@shared")
memory.semantic.add("OPC UA servers listen on TCP port 4840 by default", topic="opc-ua")
memory.semantic.ingest(
"""FactoryTalk Optix is an HMI and IIoT platform. It connects to OPC UA servers.
Optix projects publish tag data over MQTT. Use the cloud connector for remote dashboards.""",
source="optix-handbook", topic="hmi")
# Personal knowledge for one user
memory.semantic.add("Kumar's test rig uses port 4841", user_id="kumar", topic="opc-ua")
for hit in memory.semantic.search("which port does OPC UA use?", user_id="kumar"):
print(f"{hit.score:.2f} [{hit.record.user_id}] {hit.record.content}")
print("topics:", memory.semantic.topics())
# Re-ingesting a document replaces its chunks; forget_source() removes it entirely
memory.semantic.ingest("FactoryTalk Optix 2.0 adds a Python runtime.", source="optix-handbook", topic="hmi")
print("handbook now:", [r.content for r in memory.semantic.list() if (r.key or "").startswith("knowledge:optix")])
Output:
0.55 [@shared] OPC UA servers listen on TCP port 4840 by default
0.38 [kumar] Kumar's test rig uses port 4841
0.37 [@shared] FactoryTalk Optix is an HMI and IIoT platform. It connects to OPC UA servers.
0.30 [@shared] Optix projects publish tag data over MQTT. Use the cloud connector for remote dashboards.
topics: {'hmi': 2, 'opc-ua': 1}
handbook now: ['FactoryTalk Optix 2.0 adds a Python runtime.']
04 Episodic memory¶
What happened, when, where and with whom - consolidated as it ages.
examples/cookbook/04_episodic.py
import time
from kioko import MemoryLayer
DAY = 86400
memory = MemoryLayer()
user = "kumar"
memory.episodes.record("Presented the predictive-maintenance demo to the plant team", user_id=user,
where="Pune plant", who=["Krutika", "plant team"], outcome="approved for a pilot", valence=0.8)
for line in (1, 2, 3):
memory.episodes.record(f"Calibrated the vision sensor on packaging line {line}", user_id=user,
when=time.time() - (60 + line) * DAY, where="Pune plant", outcome="within tolerance")
for episode in memory.episodes.timeline(user_id=user):
print(f"{time.strftime('%Y-%m-%d', time.gmtime(episode.occurred_at))} {episode.what} -> {episode.outcome}")
print("last 7 days:", [e.what for e in memory.episodes.recent(user_id=user, days=7)])
# Old, similar episodes are folded into one summary memory (and kept, at half importance)
print(memory.episodes.consolidate(user_id=user, older_than_days=30))
print([r.content[:90] for r in memory.list(user_id=user, memory_types=["summary"])])
# Conversation turns with a time reference become episodes automatically
memory.learn("Yesterday I met Krutika to review the architecture.", user_id=user)
print([e.what for e in memory.episodes.recent(user_id=user, days=2)])
Output:
2026-10-27 Presented the predictive-maintenance demo to the plant team -> approved for a pilot
2026-08-27 Calibrated the vision sensor on packaging line 1 -> within tolerance
2026-08-26 Calibrated the vision sensor on packaging line 2 -> within tolerance
2026-08-25 Calibrated the vision sensor on packaging line 3 -> within tolerance
last 7 days: ['Presented the predictive-maintenance demo to the plant team']
{'clusters': 1, 'summaries': ['<id>'], 'consolidated': 3, 'forgotten': 0}
['3 episodes (2026-08-25 to 2026-08-27): Calibrated the vision sensor on packaging line 3; C']
['Presented the predictive-maintenance demo to the plant team', 'User met Krutika to review the architecture']
05 Working memory¶
The current task's scratchpad - per session, small, expiring, never embedded.
examples/cookbook/05_working.py
from kioko import MemoryLayer
memory = MemoryLayer()
wm = memory.working("ticket-4711", user_id="kumar", capacity=4)
wm.set("task", "Find why line 3's OPC UA client disconnects every hour")
wm.set("hypothesis", "session timeout shorter than the publishing interval")
wm.note("Server log shows BadSessionIdInvalid at 14:00, 15:00, 16:00")
wm.set("next_step", "compare the client's session timeout with the server's")
print(wm.render())
# The scratchpad is part of the agent's context for this session (and only this session)
print(memory.context("what should I check next?", user_id="kumar", session_id="ticket-4711").text)
# Keep a conclusion for the long term, drop the rest when the task is done
wm.set("finding", "Client session timeout was 60 minutes; the server closes idle sessions after 59")
fact = wm.promote("finding")
print("promoted:", fact.memory_type, "-", fact.content)
print("cleared:", wm.clear(), "items")
Output:
- task: Find why line 3's OPC UA client disconnects every hour
- hypothesis: session timeout shorter than the publishing interval
- note-138310630: Server log shows BadSessionIdInvalid at 14:00, 15:00, 16:00
- next_step: compare the client's session timeout with the server's
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
Current task notes (working memory):
- task: Find why line 3's OPC UA client disconnects every hour
- hypothesis: session timeout shorter than the publishing interval
- note-138310630: Server log shows BadSessionIdInvalid at 14:00, 15:00, 16:00
- next_step: compare the client's session timeout with the server's
</memory>
promoted: fact - Client session timeout was 60 minutes; the server closes idle sessions after 59
cleared: 3 items
06 Conversational memory¶
Rolling summary + the last few turns - constant size however long the chat gets.
examples/cookbook/06_conversation.py
from kioko import MemoryLayer
memory = MemoryLayer()
chat = memory.conversation("support-chat-1", user_id="kumar", keep_last=2, summarize_after=8)
turns = [
("The packaging line stopped with fault E-217.", "E-217 is a servo overtravel. Check the limit switches."),
("Limit switches look fine.", "Then check the encoder cable on axis 2."),
("The encoder cable was loose; I reseated it.", "Good. Clear the fault and jog the axis slowly."),
("Axis 2 jogs fine now.", "Run one cycle at 20% speed."),
("One cycle completed.", "Restore production speed and watch the following error."),
("Following error stays under 0.1 mm.", "The line is back to normal."),
]
for user_text, reply in turns:
chat.add_turn(user_text, reply)
print("stats:", chat.stats())
print("summary:", chat.summary())
print()
print(chat.context(max_tokens=200))
print()
print([m["role"] for m in chat.as_messages(system="You are a maintenance assistant.")])
Output:
stats: {'messages': 12, 'summarized': 8, 'verbatim': 4, 'summary_tokens': 41, 'window_tokens': 97}
summary: User said: The packaging line stopped with fault E-217. | Limit switches look fine. User said: The encoder cable was loose; I reseated it. | Axis 2 jogs fine now.
Earlier in this conversation:
User said: The packaging line stopped with fault E-217. | Limit switches look fine. User said: The encoder cable was loose; I reseated it. | Axis 2 jogs fine now.
Recent messages:
user: One cycle completed.
assistant: Restore production speed and watch the following error.
user: Following error stays under 0.1 mm.
assistant: The line is back to normal.
['system', 'system', 'user', 'assistant', 'user', 'assistant']
07 Goal memory¶
Objectives with priority, progress, due dates and sub-goals - always in the agent's view.
examples/cookbook/07_goals.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
release = memory.goals.add("Ship the Kioko memory layer v1", user_id=user, priority="high", due="2026-12-15",
criteria="package, docs and console published", why="agents forget everything")
memory.goals.add("Write the cookbook", user_id=user, parent=release)
memory.goals.add("Benchmark pgvector vs Qdrant", user_id=user, parent=release, priority="medium")
memory.goals.add("Fix the release blocker", user_id=user, priority="critical", due="2026-10-01")
memory.goals.update(release, user_id=user, progress=0.6, note="all memory kinds implemented")
memory.goals.complete("Write the cookbook", user_id=user, note="published with the docs")
print(memory.goals.render(user_id=user))
for node in memory.goals.tree(user_id=user):
print(node["title"], "->", [(child["title"], child["status"]) for child in node["children"]])
# Goals stated in conversation are captured too (with the due date parsed)
memory.learn("My goal is to pass the Azure AI Engineer exam by March 31.", user_id=user)
print([(g.title, g.due) for g in memory.goals.active(user_id=user) if "Azure" in g.title])
Output:
- Fix the release blocker (critical, 0%, OVERDUE since 2026-10-01)
- Ship the Kioko memory layer v1 (high, 60%, due 2026-12-15)
- Benchmark pgvector vs Qdrant (medium, 0%)
Fix the release blocker -> []
Ship the Kioko memory layer v1 -> [('Write the cookbook', 'done'), ('Benchmark pgvector vs Qdrant', 'active')]
[('Pass the Azure AI Engineer exam by March 31', '2027-03-31')]
08 Procedural memory¶
How-tos with versioned steps and a success record, plus standing rules.
examples/cookbook/08_procedural.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.procedures.add("deploy the agent to the edge gateway", user_id=user,
steps=["build the container image", "push it to the plant registry",
"roll it out with the fleet manager"],
when_to_use="releasing a new agent version", tools=["docker", "fleet-manager"])
memory.procedures.record_outcome("deploy the agent to the edge gateway", success=True, user_id=user)
# A better way is found: same name, new steps -> version 2 (success counts carry over)
v2 = memory.procedures.add("deploy the agent to the edge gateway", user_id=user,
steps=["build the container image", "run the smoke tests", "push it to the plant registry",
"roll out to one gateway, then the fleet"])
print(v2.render(), f"\n(version {v2.version}, success rate {v2.success_rate})")
for procedure, score in memory.procedures.find("how do I release a new version of the agent?", user_id=user):
print(f"{score:.2f} {procedure.name}")
# Behavioural rules are procedural memory too (they are always in the context)
memory.procedures.add_rule("Always run the smoke tests before a rollout", user_id=user)
print([r.content for r in memory.procedures.rules(user_id=user)])
# Described in conversation -> stored as a procedure
memory.learn("To reset the safety relay, switch off the main breaker, then press reset for 5 seconds, "
"then switch the breaker back on.", user_id="kumar")
print(memory.procedures.get("reset the safety relay", user_id=user).steps)
Output:
How to deploy the agent to the edge gateway (use when releasing a new agent version):
1. build the container image
2. run the smoke tests
3. push it to the plant registry
4. roll out to one gateway, then the fleet
(version 2, success rate 1.0)
0.56 deploy the agent to the edge gateway
['Always run the smoke tests before a rollout']
['Switch off the main breaker', 'Press reset for 5 seconds', 'Switch the breaker back on']
09 Skill memory¶
Reusable prompts, tools and code - picked by relevance x reliability.
examples/cookbook/09_skills.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.skills.add("summarize alarm log", user_id=user, kind="prompt",
description="Summarize a PLC alarm log into the top recurring faults with counts",
body="Group these alarms by code and list the 5 most frequent with counts:\n{alarms}")
memory.skills.add("parse alarm csv", user_id=user, kind="code", language="python",
description="Parse a FactoryTalk alarm export (CSV) into records",
body="import csv\nrows = list(csv.DictReader(open(path)))")
memory.skills.add("summarize alarm log (short)", user_id=user, kind="prompt",
description="Summarize a PLC alarm log in one line")
for _ in range(5):
memory.skills.record_use("summarize alarm log", success=True, latency_ms=900, user_id=user)
memory.skills.record_use("summarize alarm log (short)", success=False, user_id=user, note="missed the top fault")
for skill, score in memory.skills.find("summarize today's alarm log", user_id=user):
print(f"{score:.2f} {skill.name} reliability={skill.reliability:.2f} uses={skill.uses}")
print(memory.skills.render_prompt("summarize alarm log", user_id=user, alarms="E-217 x12, E-301 x4"))
Output:
0.48 summarize alarm log reliability=0.57 uses=5
0.31 summarize alarm log (short) reliability=0.00 uses=1
0.23 parse alarm csv reliability=0.50 uses=0
Group these alarms by code and list the 5 most frequent with counts:
E-217 x12, E-301 x4
10 Strategy memory¶
What worked for which kind of task (Reflexion / ExpeL-style lessons).
examples/cookbook/10_strategies.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
task = "debugging an OPC UA connection"
memory.strategies.reflect(task, "check the server certificate trust list first", "success", user_id=user,
lesson="most OPC UA connection failures are untrusted certificates")
memory.strategies.reflect(task, "check the server certificate trust list first", "success", user_id=user)
memory.strategies.reflect(task, "restart the OPC UA server", "failure", user_id=user,
lesson="restarting hides the real cause and drops every other client")
memory.strategies.reflect(task, "increase the session timeout", "partial", user_id=user)
print("recommended:")
for strategy, score in memory.strategies.recommend("my OPC UA client cannot connect", user_id=user):
print(f" {score:.2f} {strategy.line()}")
print("avoid:")
for strategy in memory.strategies.avoid("my OPC UA client cannot connect", user_id=user):
print(" ", strategy.line())
# A lesson stated in conversation becomes a strategy as well
memory.learn("Lesson learned: when commissioning a new drive, back up the parameters first.", user_id=user)
print([s.line() for s in memory.strategies.list(user_id=user) if "drive" in s.task_type])
Output:
recommended:
0.24 For debugging an OPC UA connection: check the server certificate trust list first (2/2 worked). Lesson: most OPC UA connection failures are untrusted certificates
0.15 For debugging an OPC UA connection: increase the session timeout (0/1 worked).
avoid:
For debugging an OPC UA connection: restart the OPC UA server (0/1 worked). Lesson: restarting hides the real cause and drops every other client
['For commissioning a new drive: Back up the parameters first.']
11 Decision memory¶
ADR-style decisions with context, options, rationale, outcome and supersession.
examples/cookbook/11_decisions.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
first = memory.decisions.record("vector database for production", chosen="Qdrant", user_id=user,
options=["Qdrant", "pgvector", "Azure AI Search"],
context="memory service for plant agents", rationale="fast to try",
decided_by="architecture review")
second = memory.decisions.supersede(first, chosen="pgvector", user_id=user,
rationale="we already operate Postgres; one database to back up")
memory.decisions.set_outcome(second, "running in production since the pilot", user_id=user)
for decision in memory.decisions.log(user_id=user):
print(f"[{decision.status}] {decision.line()}")
print("chain:", " -> ".join(d.chosen for d in memory.decisions.chain(second, user_id=user)))
# Superseded decisions never reach the agent's context; the current one does when relevant
print(memory.context("which vector database do we use?", user_id=user, sections=["decisions"]).text)
# From conversation
memory.learn("We decided to use MQTT for telemetry because the plant already runs a broker.", user_id=user)
print(memory.decisions.log(user_id=user)[-1].line())
Output:
[superseded] vector database for production: chose Qdrant because fast to try (superseded 2026-10-27)
[accepted] vector database for production: chose pgvector because we already operate Postgres; one database to back up (accepted 2026-10-27)
chain: Qdrant -> pgvector
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
Earlier decisions:
- vector database for production: chose pgvector because we already operate Postgres; one database to back up (accepted 2026-10-27)
</memory>
Decided: use MQTT for telemetry because the plant already runs a broker (accepted 2026-10-27)
12 Graph memory¶
A temporal knowledge graph - relations with history, multi-hop queries, entity-aware recall.
examples/cookbook/12_graph.py
import time
from kioko import MemoryLayer
memory = MemoryLayer(auto_graph=True) # also learn relations from conversations
user = "kumar"
graph = memory.graph
graph.add("Kumar", "works_at", "Contoso", user_id=user)
before_move = time.time()
time.sleep(0.01)
graph.add("Kumar", "works_at", "Rockwell Automation", user_id=user) # single-valued: replaces, keeps history
graph.add("Kumar", "reports_to", "Krutika", user_id=user)
graph.add("Krutika", "leads", "AI & IIoT architecture", user_id=user)
graph.add("AI & IIoT architecture", "part_of", "Rockwell Automation", user_id=user)
graph.add_entity("Krutika", type="person", aliases=["KK"], user_id=user)
print("now:", [r.object for r in graph.relations(user_id=user, subject="Kumar", predicate="works_at")])
print("then:", [r.object for r in graph.relations(user_id=user, subject="Kumar", predicate="works_at", at=before_move)])
print("path:", " / ".join(r.line() for r in graph.path("Kumar", "AI & IIoT architecture", user_id=user)))
print("about KK:\n" + graph.about("KK", user_id=user))
# Learned from conversation: the user's own facts and third-party relations
memory.learn("My mentor is Ravi. Ravi manages the Pune plant.", user_id=user)
print("mentioned:", graph.match_entities("Did Ravi approve the plant change?", user_id=user))
print(memory.context("Did Ravi approve the plant change?", user_id=user, sections=["graph"]).text)
Output:
now: ['Rockwell Automation']
then: ['Contoso']
path: Kumar reports to Krutika / Krutika leads AI & IIoT architecture
about KK:
- Krutika leads AI & IIoT architecture
- Kumar reports to Krutika
mentioned: ['ravi']
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
Known relationships:
- Ravi manages the Pune plant
- User's mentor is Ravi
</memory>
13 KV-cache memory¶
Keep the provider's prompt cache warm and never compute the same thing twice.
examples/cookbook/13_kv_cache.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.add("Always answer with code first", user_id=user, memory_type="instruction")
memory.profile.set("role", "senior AI engineer", user_id=user)
memory.goals.add("Ship the Kioko memory layer v1", user_id=user, priority="high")
memory.facts.add("The Pune plant runs ControlLogix controllers", user_id=user)
# 1) The stable prefix (instructions, profile, goals) is byte-identical across turns -> provider cache hits.
for _ in range(3):
prefix = memory.kv.prefix(user_id=user)
print(prefix.hash, prefix.tokens, "tokens, reused:", prefix.reused)
# 2) Request layouts with the cacheable part first.
system = memory.kv.anthropic_system("Which controllers does the plant use?", user_id=user,
base_system="You are a plant support agent.", ttl="1h")
print([("cache_control" in block, block["text"][:40]) for block in system])
messages = memory.kv.openai_messages("Which controllers does the plant use?", user_id=user,
base_system="You are a plant support agent.")
print([m["role"] for m in messages], "prompt_cache_key:", memory.kv.cache_key(user_id=user))
# 3) Embedding and context caches inside the layer.
for _ in range(3):
memory.context("Which controllers does the plant use?", user_id=user)
print(memory.kv.stats())
Output:
50c2f5df793fa115 42 tokens, reused: False
50c2f5df793fa115 42 tokens, reused: True
50c2f5df793fa115 42 tokens, reused: True
[(True, 'You are a plant support agent.\n\n<memory-'), (False, '<memory>\nLong-term memory about the user')]
['system', 'system', 'user'] prompt_cache_key: kioko-50c2f5df793fa115
{'prefix': {'reused': 5, 'changed': 1, 'reuse_rate': 0.833}, 'embeddings': {'size': 5, 'maxsize': 4096, 'hits': 1, 'misses': 5, 'evictions': 0, 'hit_rate': 0.167}, 'contexts': {'size': 2, 'maxsize': 256, 'hits': 3, 'misses': 2, 'evictions': 0, 'hit_rate': 0.6}}
14 Parametric memory¶
Hand stable memories over to model weights (fine-tune / LoRA) and stop paying tokens for them.
examples/cookbook/14_parametric.py
import json
import os
import tempfile
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.profile.set("location", "Pune", user_id=user)
memory.profile.set("employer", "Rockwell Automation", user_id=user)
memory.procedures.add("reset the safety relay", user_id=user,
steps=["switch off the main breaker", "press reset for 5 seconds", "switch the breaker back on"])
memory.decisions.record("telemetry protocol", chosen="MQTT", rationale="the plant already runs a broker",
user_id=user)
path = os.path.join(tempfile.mkdtemp(), "kumar-train.jsonl")
rows = memory.parametric.export_dataset(path, user_id=user)
print(f"{len(rows)} examples written to {os.path.basename(path)}")
print(json.dumps(rows[0]["messages"][1:], indent=None))
# After training, record which memories the model absorbed ...
model = "ft:gpt-4.1-mini:kioko-kumar-v1"
memory.parametric.register_model(model, base_model="gpt-4.1-mini", kind="fine-tune", dataset=path, user_id=user)
memory.parametric.mark_consolidated(rows, model=model)
# ... and requests to that model no longer carry them in the prompt.
question = "How do I reset the safety relay?"
print("base model context tokens:", memory.context(question, user_id=user).tokens)
print("fine-tuned context tokens:", memory.context(question, user_id=user, model=model).tokens)
# probe(): ask a model what it already knows (heuristic word overlap; mark_as records the hits)
def base_model(question):
return {"Where do I live?": "You live in Pune."}.get(question, "I don't know.")
print([(r["question"], r["knows"]) for r in memory.parametric.probe(base_model, user_id=user)])
Output:
4 examples written to kumar-train.jsonl
[{"role": "user", "content": "What did we decide about telemetry protocol?"}, {"role": "assistant", "content": "We chose MQTT because the plant already runs a broker."}]
base model context tokens: 106
fine-tuned context tokens: 0
[('Where do I live?', True), ('Where do I work?', False), ('How do I reset the safety relay?', False), ('What did we decide about telemetry protocol?', False)]
15 Hybrid memory¶
One retrieval over every kind and representation (dense + BM25 + graph + recency, RRF), packed into a token budget.
examples/cookbook/15_hybrid_recall.py
from kioko import MemoryLayer
memory = MemoryLayer()
user = "kumar"
memory.add("Always answer with code first", user_id=user, memory_type="instruction")
memory.profile.set("employer", "Rockwell Automation", user_id=user)
memory.goals.add("Stabilise line 3", user_id=user, priority="high")
memory.facts.add("Line 3 uses a CompactLogix 5380 with firmware 33", user_id=user)
memory.semantic.add("Firmware 33 fixes the OPC UA session leak (CVE-free maintenance release)", topic="firmware")
memory.procedures.add("update CompactLogix firmware", user_id=user,
steps=["back up the project", "flash with ControlFLASH Plus", "verify the checksum"])
memory.strategies.reflect("OPC UA disconnects", "compare client and server session timeouts", "success", user_id=user)
memory.decisions.record("firmware policy", chosen="stay on long-term-support releases", user_id=user)
memory.graph.add("Line 3", "maintained_by", "Ravi", user_id=user)
memory.episodes.record("Line 3 stopped twice because of OPC UA disconnects", user_id=user)
query = "Line 3 keeps losing its OPC UA connection - what should Ravi try first?"
ctx = memory.recall(query, user_id=user, budget_tokens=400)
print(ctx.text)
print("\nsections:", ctx.counts(), "| tokens:", ctx.tokens, "| left out:", [o["section"] for o in ctx.omitted])
print("\nwhy (signals per retriever):")
for result in ctx.relevant + ctx.knowledge + ctx.procedures + ctx.strategies:
signals = {k: v for k, v in result.signals.items() if k in ("dense", "sparse", "graph", "recency", "rrf")}
print(f" {result.score:.2f} {result.record.content[:60]:<60} {signals}")
Output:
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
Instructions from the user:
- Always answer with code first
About the user:
- User works at Rockwell Automation.
The user's open goals:
- Stabilise line 3 (high, 0%)
Relevant memories:
- Line 3 stopped twice because of OPC UA disconnects.
- Line 3 uses a CompactLogix 5380 with firmware 33
Background knowledge:
- Firmware 33 fixes the OPC UA session leak (CVE-free maintenance release)
Known relationships:
- Line 3 maintained by Ravi
Procedures that may apply:
- update CompactLogix firmware: 1) back up the project; 2) flash with ControlFLASH Plus; 3) verify the checksum
Lessons from earlier attempts:
- For OPC UA disconnects: compare client and server session timeouts (1/1 worked).
</memory>
sections: {'instructions': 1, 'profile': 1, 'goals': 1, 'relevant': 2, 'knowledge': 1, 'graph': 1, 'procedures': 1, 'strategies': 1} | tokens: 225 | left out: []
why (signals per retriever):
1.00 Line 3 stopped twice because of OPC UA disconnects. {'dense': 0.4407, 'sparse': 1.0, 'graph': 1.0, 'rrf': 0.0459}
0.98 Line 3 uses a CompactLogix 5380 with firmware 33 {'dense': 0.2605, 'sparse': 0.58, 'graph': 1.0, 'rrf': 0.04516}
0.69 Firmware 33 fixes the OPC UA session leak (CVE-free maintena {'dense': 0.2202, 'sparse': 0.3784, 'rrf': 0.0315}
0.34 How to update CompactLogix firmware: 1. back up the project. {'dense': 0.128, 'sparse': 0.1759, 'rrf': 0.01538}
0.69 For OPC UA disconnects: compare client and server session ti {'dense': 0.2122, 'sparse': 0.4269, 'rrf': 0.0315}
Keeping memory healthy¶
16 Maintenance¶
Keep memory small and sharp - summarise, consolidate, de-duplicate and forget.
examples/cookbook/16_consolidation.py
import time
from kioko import MemoryLayer
DAY = 86400
memory = MemoryLayer()
user = "kumar"
memory.add("User likes green tea", user_id=user, dedupe=False) # duplicates that slipped in
memory.add("User likes green tea", user_id=user, dedupe=False)
for i in range(3):
memory.episodes.record(f"Weekly sync with the plant team, week {i + 1}", user_id=user,
when=time.time() - (90 - 7 * i) * DAY)
faded = memory.episodes.record("Small talk about the weather", user_id=user, importance=0.05)
record = memory.get(faded.id)
record.updated_at = record.created_at = time.time() - 365 * DAY # a year old, never recalled
memory.store.update(record)
for i in range(25):
memory.add_message("user" if i % 2 == 0 else "assistant", f"message {i} about the line 3 retrofit",
session_id="retrofit", user_id=user)
print("preview:", memory.consolidate(user_id=user, dry_run=True))
report = memory.consolidate(user_id=user)
print("done: ", {k: v for k, v in report.items() if v and k != "duration_ms"})
print("kinds now:", memory.kind_counts(user))
# Run it nightly: kioko consolidate --db memory.db (or POST /v1/consolidate, or the console's button)
Output:
preview: {'users': 1, 'expired': 0, 'sessions_summarized': 0, 'episode_summaries': 0, 'episodes_consolidated': 0, 'duplicates_merged': 1, 'forgotten': 1, 'dry_run': True, 'duration_ms': <n>}
done: {'users': 1, 'sessions_summarized': 1, 'episode_summaries': 1, 'episodes_consolidated': 3, 'duplicates_merged': 1, 'forgotten': 1}
kinds now: {'episodic': 3, 'fact': 1, 'summary': 2}
Stores and the service¶
17 Database plugins¶
The same code on SQLite, Postgres + pgvector and Qdrant.
Runs against a real server when KIOKO_PG_DSN / KIOKO_QDRANT_URL are set; otherwise it uses SQLite and shows the connection strings. Start servers with: docker compose -f docker-compose.stores.yml up -d
examples/cookbook/17_stores.py
import os
import tempfile
from kioko import MemoryLayer, SQLiteStore, connect
def exercise(memory, label):
user = "store-demo"
memory.goals.add("Ship v1", user_id=user)
memory.facts.add("Line 3 uses a CompactLogix 5380", user_id=user)
memory.graph.add("Line 3", "maintained_by", "Ravi", user_id=user)
ctx = memory.recall("Who maintains line 3 and what controller does it use?", user_id=user, budget_tokens=300)
print(f"{label:<10} {ctx.counts()}")
memory.clear(user_id=user)
exercise(MemoryLayer.sqlite(os.path.join(tempfile.mkdtemp(), "memory.db")), "sqlite")
if os.environ.get("KIOKO_PG_DSN"):
exercise(connect(os.environ["KIOKO_PG_DSN"]), "pgvector") # pip install "kioko[pgvector]"
else:
print("pgvector set KIOKO_PG_DSN=postgresql://kioko:<password>@localhost:5432/kioko to run this part")
if os.environ.get("KIOKO_QDRANT_URL"):
from kioko.stores import QdrantStore
store = QdrantStore(os.environ["KIOKO_QDRANT_URL"], collection="kioko_cookbook",
api_key=os.environ.get("QDRANT_API_KEY"))
exercise(MemoryLayer(store, history_store=SQLiteStore(os.path.join(tempfile.mkdtemp(), "history.db"))), "qdrant")
else:
print("qdrant set KIOKO_QDRANT_URL=http://localhost:6333 to run this part")
Output:
sqlite {'goals': 1, 'relevant': 1, 'graph': 1}
pgvector set KIOKO_PG_DSN=postgresql://kioko:<password>@localhost:<port>/kioko to run this part
qdrant set KIOKO_QDRANT_URL=http://localhost:<port> to run this part
18 The memory service¶
The same memory over HTTP for any language (REST) and for Python (RemoteMemory).
examples/cookbook/18_rest_client.py
import json
import threading
import urllib.request
from kioko import MemoryLayer, RemoteMemory
from kioko.service.http import make_server
server = make_server(MemoryLayer(), port=0) # in production: kioko serve --store postgresql://...
threading.Thread(target=server.serve_forever, daemon=True).start()
base = f"http://127.0.0.1:{server.server_address[1]}"
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
def post(path, body):
request = urllib.request.Request(base + path, data=json.dumps(body).encode(), method="POST",
headers={"Content-Type": "application/json"})
with opener.open(request) as response:
return json.load(response)
# Any language: typed memory operations and a budgeted recall, as JSON
post("/v1/kinds/goals/add", {"user_id": "kumar", "title": "Stabilise line 3", "priority": "high"})
post("/v1/kinds/decisions/record", {"user_id": "kumar", "title": "telemetry", "chosen": "MQTT"})
post("/v1/turns", {"user_id": "kumar", "user_message": "I work at Rockwell Automation in Pune.",
"assistant_message": "Noted."})
ctx = post("/v1/recall", {"user_id": "kumar", "query": "What telemetry protocol do we use?", "budget_tokens": 300})
print(ctx["text"])
# Python: RemoteMemory has the MemoryLayer API, kinds included
remote = RemoteMemory(base)
remote.goals.update("Stabilise line 3", user_id="kumar", progress=0.5)
print([g.line() for g in remote.goals.active(user_id="kumar")])
print(remote.kind_counts("kumar"))
server.shutdown()
Output:
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
About the user:
- User works at Rockwell Automation in Pune.
The user's open goals:
- Stabilise line 3 (high, 0%)
Earlier decisions:
- telemetry: chose MQTT (accepted 2026-10-27)
</memory>
['Stabilise line 3 (high, 50%)']
{'decision': 1, 'fact': 1, 'goal': 1}
Putting it together¶
19 A complete agent turn with Kioko¶
Recall -> LLM -> learn. Swap llm for OpenAI, Azure OpenAI, Anthropic, Agno, LangChain ... (see docs/guide.md); this stand-in model runs offline.
examples/cookbook/19_agent_loop.py
from kioko import MemoryLayer
memory = MemoryLayer.sqlite(":memory:", auto_graph=True)
def llm(system: str, user: str) -> str:
"""Stand-in for a chat model: it just shows what the model would receive."""
lines = [line for line in system.splitlines() if line.startswith("- ")]
return f"(model saw {len(lines)} memory lines) " + (lines[0][2:] if lines else "Nice to meet you!")
def agent(message: str, *, user_id: str, session_id: str) -> str:
ctx = memory.recall(message, user_id=user_id, session_id=session_id, budget_tokens=800) # 1. remember
reply = llm(system="You are a plant support agent.\n\n" + ctx.text, user=message) # 2. answer
memory.record_turn(message, reply, user_id=user_id, session_id=session_id) # 3. learn
return reply
for text in ["Hi, I'm Kumar. I work at Rockwell Automation and my manager is Krutika.",
"My goal is to stabilise line 3 by November 30.",
"We decided to use MQTT for telemetry because the plant already runs a broker.",
"To reset the safety relay, switch off the breaker, then press reset, then switch it back on.",
"What do you know about me?"]:
print("you >", text)
print("agent>", agent(text, user_id="kumar", session_id="s1"))
print("\nremembered:", memory.kind_counts("kumar"))
print(memory.recall("how do I reset the safety relay?", user_id="kumar").text)
Output:
you > Hi, I'm Kumar. I work at Rockwell Automation and my manager is Krutika.
agent> (model saw 0 memory lines) Nice to meet you!
you > My goal is to stabilise line 3 by November 30.
agent> (model saw 3 memory lines) User's manager is Krutika.
you > We decided to use MQTT for telemetry because the plant already runs a broker.
agent> (model saw 4 memory lines) User's manager is Krutika.
you > To reset the safety relay, switch off the breaker, then press reset, then switch it back on.
agent> (model saw 4 memory lines) User's manager is Krutika.
you > What do you know about me?
agent> (model saw 4 memory lines) User's manager is Krutika.
remembered: {'decision': 1, 'fact': 3, 'goal': 1, 'procedure': 1, 'relation': 3}
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
About the user:
- User's manager is Krutika.
- User works at Rockwell Automation.
- User's name is Kumar.
The user's open goals:
- Stabilise line 3 by November 30 (medium, 0%, due 2026-11-30)
Procedures that may apply:
- reset the safety relay: 1) Switch off the breaker; 2) Press reset; 3) Switch it back on
</memory>
20 Self-managing memory in one line¶
Kioko.intelligent() decides what each turn means for memory - which kind it belongs to and which operation to apply - plans what to retrieve per question, and tidies up on its own. Offline (rules + algorithms); pass llm= to let a model decide (see 21_llm_intelligence.py).
examples/cookbook/20_intelligent_memory.py
import kioko
memory = kioko.intelligent() # or kioko.intelligent("sqlite:///memory.db"), "postgresql://...", ...
user, session = "kumar", "shift-1"
# 1. Preview what the intelligence would do with a message - nothing is written in a dry run.
preview = memory.manage("I'm Kumar, I live in Pune. My goal is to stabilise line 3 by November 30.",
user_id=user, dry_run=True)
print(preview.summary())
# 2. A real conversation. In an agent, this is the record_turn() you already call after each reply.
turns = [
"I'm Kumar, I live in Pune. My goal is to stabilise line 3 by November 30.",
"We decided to use Qdrant for the maintenance notes because it is fast.",
"I'm debugging ticket INC-4521, the drive shows error E-217 on line 3.",
"To reset the safety relay, switch off the breaker, then press reset for 5 seconds, then switch it back on.",
"We switched from Qdrant to pgvector because we already run Postgres.",
"I'm about 60% done with stabilising line 3.",
"thanks!",
]
for text in turns:
report = memory.manage(text, user_id=user, session_id=session)
print(f"\nyou > {text}")
for line in report.summary().splitlines():
print(" " + line)
# 3. Every kind was updated where it belongs.
goal = memory.goals.active(user_id=user)[0]
print("\ngoal:", goal.line())
print("decision:", [(d.title, d.chosen, d.status) for d in memory.decisions.log(user_id=user)])
print("working memory:", memory.working(session, user_id=user).snapshot())
# 4. Retrieval is planned by intent: a how-to question gets the procedure first; small talk costs nothing.
ctx = memory.recall("How do I reset the safety relay?", user_id=user, session_id=session)
print("\nplan:", ctx.plan["intents"], "-", ctx.plan["reason"])
print(ctx.text)
# 5. Feedback closes the loop: "that worked" is credited to the procedure the agent just used.
memory.record_turn("How do I reset the safety relay?", "Switch off the breaker, press reset, switch it on.",
user_id=user, session_id=session)
memory.recall("That worked, thanks!", user_id=user, session_id=session)
memory.record_turn("That worked, thanks!", "Great!", user_id=user, session_id=session)
procedure = memory.procedures.get("reset the safety relay", user_id=user)
print("procedure track record:", procedure.successes, "success,", procedure.failures, "failures")
# 6. Done with the shift: what lasts from the scratchpad becomes an episode, the session is summarised.
memory.working(session, user_id=user).set("root_cause", "a loose encoder cable")
print("\nend of session:", memory.end_session(session, user_id=user)["result"])
print("intelligence stats:", {k: v for k, v in memory.intelligence.stats().items() if k in ("turns", "gated")})
Output:
[dry run] profile.set name = Kumar
[dry run] profile.set location = Pune
[dry run] goal.add "Goal: Stabilise line 3 by November 30."
[dry run] relation.add (user) -named-> (Kumar)
[dry run] relation.add (user) -lives_in-> (Pune)
you > I'm Kumar, I live in Pune. My goal is to stabilise line 3 by November 30.
profile.set name = Kumar
profile.set location = Pune
goal.add "Goal: Stabilise line 3 by November 30."
relation.add (user) -named-> (Kumar)
relation.add (user) -lives_in-> (Pune)
you > We decided to use Qdrant for the maintenance notes because it is fast.
decision.add "Decision: use Qdrant for the maintenance notes because it is fast."
you > I'm debugging ticket INC-4521, the drive shows error E-217 on line 3.
working.set ticket = INC-4521
working.set error = E-217
working.set task = Debugging ticket INC-4521, the drive shows error E-217 on line 3
you > To reset the safety relay, switch off the breaker, then press reset for 5 seconds, then switch it back on.
procedure.add "How to reset the safety relay: 1. Switch off the breaker. 2. Press re…"
you > We switched from Qdrant to pgvector because we already run Postgres.
decision.supersede "use Qdrant for the maintenance notes" -> pgvector
linked: decision "Qdrant vs pgvector - chose pgvector because we already run…" is related to decision "use Qdrant for the maintenance notes because…
linked: decision "Qdrant vs pgvector - chose pgvector because we already run…" supersedes decision "use Qdrant for the maintenance notes because it…
you > I'm about 60% done with stabilising line 3.
goal.progress "Stabilise line 3 by November 30" -> 60%
you > thanks!
nothing to remember (small talk)
goal: Stabilise line 3 by November 30 (medium, 60%, due 2026-11-30)
decision: [('use Qdrant for the maintenance notes', 'use Qdrant for the maintenance notes', 'superseded'), ('Qdrant vs pgvector', 'pgvector', 'accepted')]
working memory: {'ticket': 'INC-4521', 'error': 'E-217', 'task': 'Debugging ticket INC-4521, the drive shows error E-217 on line 3'}
plan: ['howto'] - a how-to question: procedures, skills and lessons first
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
About the user:
- User lives in Pune.
... (17 more lines)
21 Model-driven memory management¶
One LLM call per turn decides every memory operation, and Kioko's algorithms skip trivial turns (no call), validate what the model proposes, and take over if the model fails.
Any prompt -> text function works. With OpenAI / Azure OpenAI it is a few lines:
from openai import OpenAI # or AzureOpenAI(azure_endpoint=..., api_version=...)
client = OpenAI()
def ask(prompt: str) -> str:
reply = client.chat.completions.create(model=os.environ["KIOKO_MODEL"], temperature=0,
response_format={"type": "json_object"},
messages=[{"role": "user", "content": prompt}])
return reply.choices[0].message.content
memory = kioko.intelligent("sqlite:///memory.db", llm=ask)
An Agno agent (llm=Agent(model=...)) or a LangChain chat model works the same way. This recipe uses a scripted stand-in model so it runs offline and prints the same thing every time.
examples/cookbook/21_llm_intelligence.py
import json
import kioko
SCRIPT = { # what our stand-in "model" answers for each user message
"I'm Kumar from the Pune plant. Our team wants to cut unplanned downtime on line 3 by 20% this year.": [
{"op": "profile.set", "slot": "location", "value": "Pune", "reason": "where the user works"},
{"op": "goal.add", "title": "Cut unplanned downtime on line 3 by 20%", "due": "2026-12-31",
"priority": "high", "reason": "a team objective with a target"}],
"After a long debate we picked Grafana over Power BI for the line dashboards, mostly for the alerting.": [
{"op": "decision.add", "title": "Dashboard tool for line 3", "chosen": "Grafana",
"options": ["Grafana", "Power BI"], "rationale": "better alerting", "reason": "a decision with a reason"}],
"Downtime is already down 12%.": [
{"op": "goal.progress", "target": "g1", "progress": 0.6, "note": "12% of the 20% target",
"reason": "12 of 20 percentage points"}],
"My API key is sk-EXAMPLE-not-a-real-key-000, keep it for later.": [
{"op": "fact.add", "content": "User's API key is sk-EXAMPLE-not-a-real-key-000"}],
}
FAILS_ON = "I prefer bullet points over long paragraphs." # the stand-in model "fails" on this one
def stand_in_model(prompt: str) -> str:
"""Pretends to be an LLM: finds the user's message in the prompt and answers with scripted JSON."""
said = prompt.rsplit("Conversation:", 1)[1].split("user: ", 1)[1].split("\n", 1)[0].strip()
if said == FAILS_ON:
return "Sorry, I can't produce JSON right now." # a failing model: the offline router takes over
return json.dumps({"operations": SCRIPT.get(said, [])})
calls = []
memory = kioko.intelligent(llm=lambda prompt: calls.append(prompt) or stand_in_model(prompt))
user = "kumar"
for text in [*SCRIPT, "thanks!", FAILS_ON]:
report = memory.manage(text, user_id=user)
print(f"\nyou > {text}\n [{report.router}, {report.llm_calls} LLM call(s)]")
for op in report.ops:
note = "" if op.status == "applied" else f" <- {op.status}: {op.error}"
print(f" {op.describe()}{note}")
if report.gated:
print(f" skipped before calling the model: {report.gated}")
if report.fallback:
print(f" the model failed ({report.fallback}); the offline router decided instead")
print("\nwhat the model was shown for 'Downtime is already down 12%.':")
prompt = next(p for p in calls if "Downtime is already down 12%." in p)
print("\n".join(" " + line for line in prompt.split("Open goals:")[1].split("Current decisions:")[0].strip().splitlines()))
print("\ngoal now:", memory.goals.active(user_id=user)[0].line())
print("model calls:", len(calls), "for", memory.intelligence.stats()["turns"], "turns")
Output:
you > I'm Kumar from the Pune plant. Our team wants to cut unplanned downtime on line 3 by 20% this year.
[llm, 1 LLM call(s)]
profile.set location = Pune
goal.add "Cut unplanned downtime on line 3 by 20%"
you > After a long debate we picked Grafana over Power BI for the line dashboards, mostly for the alerting.
[llm, 1 LLM call(s)]
decision.add "Dashboard tool for line 3" -> Grafana
you > Downtime is already down 12%.
[llm, 1 LLM call(s)]
goal.progress "Cut unplanned downtime on line 3 by 20%" -> 60%
you > My API key is sk-EXAMPLE-not-a-real-key-000, keep it for later.
[llm, 1 LLM call(s)]
fact.add "User's API key is sk-EXAMPLE-not-a-real-key-000" <- skipped: rejected by a processor (for example it looked like a secret)
you > thanks!
[llm, 0 LLM call(s)]
skipped before calling the model: small talk
you > I prefer bullet points over long paragraphs.
[llm->heuristic, 1 LLM call(s)]
instruction.add "User prefers bullet points over long paragraphs."
the model failed (ValueError: no operations JSON found in the model output); the offline router decided instead
what the model was shown for 'Downtime is already down 12%.':
g1 Cut unplanned downtime on line 3 by 20% (high, 0%, due 2026-12-31)
goal now: Cut unplanned downtime on line 3 by 20% (high, 60%, due 2026-12-31)
model calls: 5 for 6 turns
22 How much the intelligence saves¶
Tokens per recall with and without a plan, model calls skipped by the gate, and what rides in the cacheable prefix. Runs offline; the numbers are measured, not assumed.
examples/cookbook/22_efficiency.py
import json
import math
import kioko
from kioko import FunctionEmbedder, HashingEmbedder
# A stand-in for a semantic embedding model. Like real ones, it gives unrelated texts a baseline similarity, so
# every memory competes for a place in the context - the situation in which the plan's caps matter. (With the
# default lexical embedder retrieval is already narrow and planning mostly changes *what* is retrieved.)
_lexical = HashingEmbedder()
def semantic_like(text):
vector = [0.8 * x for x in _lexical.embed_one(text)]
vector[0] += 0.6
norm = math.sqrt(sum(x * x for x in vector))
return [x / norm for x in vector]
memory = kioko.intelligent(embedder=FunctionEmbedder(semantic_like, batch=False))
user = "kumar"
# -- a realistic memory: profile, rules, goals, decisions, procedures, lessons, facts, episodes, knowledge ---------
memory.profile.set("name", "Kumar", user_id=user)
memory.profile.set("location", "Pune", user_id=user)
memory.profile.set("role", "senior AI engineer", user_id=user)
memory.procedures.add_rule("Always cite the ticket number", user_id=user)
memory.goals.add("Stabilise line 3", user_id=user, priority="high", due="2026-11-30")
memory.goals.add("Pass the Azure AI Engineer exam", user_id=user, due="2027-03-31")
for title, chosen, why in [("Vector store", "pgvector", "we already run Postgres"),
("Telemetry protocol", "MQTT", "the plant runs a broker"),
("Dashboards", "Grafana", "better alerting"),
("Historian", "InfluxDB", "retention policies")]:
memory.decisions.record(title, chosen=chosen, rationale=why, user_id=user)
for name, steps in [("Reset the safety relay", ["Switch off the breaker", "Press reset for 5 s", "Switch on"]),
("Restart the OPC UA server", ["Stop the service", "Clear the cache", "Start the service"]),
("Recalibrate the line 3 encoder", ["Stop the line", "Run the wizard", "Verify 10 cycles"]),
("Deploy the agent", ["Run the tests", "Build the image", "Roll out to staging"])]:
memory.procedures.add(name, steps=steps, user_id=user)
memory.strategies.reflect("debugging drives", "check the encoder cable first", "success", user_id=user)
memory.strategies.reflect("debugging drives", "replace the drive", "failure", user_id=user,
lesson="expensive and rarely the cause")
for i in range(40):
memory.add(f"Line {i % 8 + 1} station {i} uses a {['Kinetix', 'PowerFlex', 'ControlLogix', 'Micro850'][i % 4]} "
f"device with firmware {i % 5}.{i % 3}", user_id=user)
for i in range(30):
memory.episodes.record(f"Shift {i}: line {i % 8 + 1} stopped for {5 + i % 20} minutes ({['jam', 'fault E-217', 'operator break', 'changeover'][i % 4]})",
user_id=user)
for i in range(20):
memory.semantic.add(f"Knowledge item {i}: OPC UA endpoint {i} listens on port {4840 + i}", user_id=user)
# -- 1. tokens per recall: every kind with default caps vs. the intent-based plan --------------------------------------
queries = ["How do I reset the safety relay?", "Why did we choose pgvector?", "What are my goals?",
"What do you know about me?", "thanks!", "When did line 3 stop for a jam?",
"Which device does station 12 use?", "What is the port of OPC UA endpoint 3?"]
rows = []
for query in queries:
full = memory.recall(query, user_id=user, plan=False, budget_tokens=None, use_cache=False)
planned = memory.recall(query, user_id=user, plan="auto", budget_tokens=None, use_cache=False)
rows.append((query, full.tokens, planned.tokens, ",".join(planned.plan["intents"]) or "general"))
print(f"{'query':<40} {'every kind':>10} {'planned':>8} intents")
for query, a, b, intents in rows:
print(f"{query:<40} {a:>10} {b:>8} {intents}")
total_full, total_planned = sum(r[1] for r in rows), sum(r[2] for r in rows)
print(f"{'total':<40} {total_full:>10} {total_planned:>8} -> {100 * (1 - total_planned / total_full):.0f}% fewer tokens")
# -- 2. model calls the gate avoids --------------------------------------------------------------------------------------
calls = []
smart = kioko.intelligent(llm=lambda prompt: calls.append(prompt) or json.dumps({"operations": []}))
conversation = ["Hi!", "How do I reset the safety relay?", "ok", "That worked, thanks!",
"We moved the line 3 changeover to Fridays.", "What did we decide about dashboards?", "cool",
"Can you write the shift report?", "The encoder on station 12 was loose.", "thanks!",
"My goal is to cut downtime by 20%.", "great", "What are my goals?", "bye"]
for text in conversation:
smart.manage(text, user_id=user)
print(f"\ngate: {len(calls)} model calls for {len(conversation)} turns "
f"({len(conversation) - len(calls)} skipped as small talk, questions or requests)")
# -- 3. hot memories join the cacheable prefix ------------------------------------------------------------------------------
team = kioko.intelligent()
team.profile.set("name", "Kumar", user_id=user)
subnet = team.facts.add("The plant's PLC network uses the 10.20.0.0/16 subnet", user_id=user)
team.facts.add("The canteen closes at 3 pm", user_id=user)
for question in ["Which subnet is the PLC network on?", "Is 10.20.4.7 inside the PLC subnet?",
"What IP range should the new PLC get?", "Can the historian reach the PLC network subnet?",
"Which subnet do the line 3 PLCs use?"]:
team.recall(question, user_id=user)
print("\nmaintenance:", team.maintain(user_id=user, tasks=["pin"]))
first = team.kv.prefix(user_id=user)
second = team.kv.prefix(user_id=user)
print(f"stable prefix: {first.tokens} tokens ({', '.join(first.sections)}); byte-identical next turn: {second.reused}")
print(first.text)
Output:
query every kind planned intents
How do I reset the safety relay? 736 611 howto
Why did we choose pgvector? 732 657 decision
What are my goals? 648 502 goals
What do you know about me? 239 239 personal
thanks! 656 123 smalltalk
When did line 3 stop for a jam? 540 540 temporal
Which device does station 12 use? 675 675 general
What is the port of OPC UA endpoint 3? 616 569 knowledge
total 4842 3916 -> 19% fewer tokens
gate: 4 model calls for 14 turns (10 skipped as small talk, questions or requests)
maintenance: [{'task': 'pin', 'pinned': ["The plant's PLC network uses the 10.20.0.0/16 subnet"], 'unpinned': [], 'result': '1 pinned, 0 unpinned'}]
stable prefix: 33 tokens (profile, pinned); byte-identical next turn: True
About the user:
- User's name is Kumar.
Keep in mind (frequently needed):
- The plant's PLC network uses the 10.20.0.0/16 subnet
23 Memory graph¶
Every memory is linked to the memories it relates to - a decision serves a goal, a fix resolves an incident, a lesson refines a procedure - with the evidence for each link. Recall follows the links and tells the agent how things connect. The algorithm is described in docs/graph-memory.md.
examples/cookbook/23_memory_graph.py
import kioko
memory = kioko.intelligent() # linking is on with the intelligence (or MemoryLayer(auto_link=True))
user = "kumar"
# 1. A goal, and one fact in the entity graph: Kioko uses pgvector.
goal = memory.goals.add("Ship the Kioko memory layer by December 15", user_id=user)
memory.graph.add("Kioko", "uses", "pgvector", user_id=user)
# 2. A decision mentions pgvector, not Kioko - the entity graph bridges them.
report = memory.manage("We decided to use pgvector because we already run Postgres.", user_id=user)
print(report.summary())
# 3. An incident: a fault, then its fix, in one session. Neither says "resolves"; the algorithm sees a problem and
# a fix about the same asset in the same session.
memory.procedures.add("Clear a conveyor jam on line 3", user_id=user,
steps=["Press stop", "Open the guard", "Remove the jammed bottle", "Reset the drive"])
for text in ("The line 3 conveyor jammed twice this morning.", "Line 3 keeps faulting with error E-217.",
"Found it: the encoder cable on line 3 was loose."):
memory.manage(text, user_id=user, session_id="shift-1")
memory.strategies.reflect("clearing conveyor jams", "reset the drive before reopening the guard", "success",
lesson="always check the photo-eye alignment after a jam", user_id=user)
# 4. Every link carries its type, weight, author and evidence.
print("\nlinks of the goal:")
for link in memory.graph.links(goal.id):
print(f" {link.line()} [{link.weight:.2f} {link.by}] {link.why}")
fault = next(r for r in memory.list(user_id=user, memory_types=["episodic"]) if "E-217" in r.content)
print("\nlinks of the fault:")
for link in memory.graph.links(fault.id):
print(f" {link.line()} [{link.weight:.2f} {link.by}] {link.why}")
# 5. Why are two memories (not) linked? explain() shows the evidence the algorithm used.
tea = memory.add("User likes green tea.", user_id=user)
verdict = memory.graph.explain(tea.id, fault.id)
print("\ntea vs fault:", verdict.linked, {k: round(v, 2) for k, v in verdict.signals.items() if v})
# 6. Spreading activation: what is connected to the lesson, best first, and through which links.
lesson = memory.strategies.list(user_id=user)[0]
print("\nrelated to the lesson:")
for item in memory.graph.related(lesson.id, depth=2, limit=4):
print(f" {item.score:.2f} {item.record.content[:60]}")
for step in item.path:
print(f" via {step.line()}")
# 7. Recall follows the links and says how the memories connect.
ctx = memory.recall("Line 3 is faulting with E-217 again - what do we know?", user_id=user)
print()
print(ctx.text)
print("brought in by links:", [r.record.content[:40] for section in ("procedures", "strategies", "relevant")
for r in getattr(ctx, section) if "linked" in r.signals])
# 8. Link by hand (never fades), trace a chain, and draw the graph.
link = memory.graph.link(lesson.id, fault.id, type="learned_from", why="the E-217 night taught us this")
print("\nmanual:", link.line(), link.by)
print("trace lesson -> fault:", " | ".join(step.line() for step in memory.graph.trace(lesson.id, fault.id)))
graph = memory.graph.memories(user_id=user)
print("graph:", graph["stats"])
# 9. Memories written before linking was on are linked in one call.
legacy = kioko.MemoryLayer(auto_link=False)
legacy.add("The OPC UA server on the HMI panel rejected our client certificate.", user_id=user,
memory_type="episodic")
legacy.strategies.reflect("debugging OPC UA", "restart the server", "failure",
lesson="check the OPC UA certificate trust list first", user_id=user)
print("\nconnect():", {k: v for k, v in legacy.graph.connect(user_id=user).items() if k != "duration_ms"})
Output:
decision.add "Decision: use pgvector because we already run Postgres."
linked: decision "use pgvector because we already run Postgres" serves goal "Ship the Kioko memory layer by December 15"
links of the goal:
decision "use pgvector because we already run Postgres" serves goal "Ship the Kioko memory layer by December 15" [0.60 auto] via Kioko uses pgvector
links of the fault:
event "Found it: the encoder cable on line 3 was loose" resolves event "Line 3 keeps faulting with error E-217" [0.67 auto] shared: line 3; similar meaning (0.30); a problem and its fix; same session, 0 min apart
event "Line 3 keeps faulting with error E-217" follows event "The line 3 conveyor jammed twice this morning" [0.50 auto] shared: Line 3; similar meaning (0.30); same session, 0 min apart
tea vs fault: False {'prior': 0.75}
related to the lesson:
0.35 The line 3 conveyor jammed twice this morning.
via lesson "For clearing conveyor jams: reset the drive before reopenin…" was learned from event "The line 3 conveyor jammed twice this morning"
0.34 How to Clear a conveyor jam on line 3: 1. Press stop. 2. Ope
via lesson "For clearing conveyor jams: reset the drive before reopenin…" refines procedure "Clear a conveyor jam on line 3"
0.05 Found it: the encoder cable on line 3 was loose.
via lesson "For clearing conveyor jams: reset the drive before reopenin…" was learned from event "The line 3 conveyor jammed twice this morning"
via event "Found it: the encoder cable on line 3 was loose" resolves event "The line 3 conveyor jammed twice this morning"
0.04 Line 3 keeps faulting with error E-217.
via lesson "For clearing conveyor jams: reset the drive before reopenin…" was learned from event "The line 3 conveyor jammed twice this morning"
via event "Line 3 keeps faulting with error E-217" follows event "The line 3 conveyor jammed twice this morning"
<memory>
Long-term memory about the user. Use it to personalise your answer. If it conflicts with the user's latest message, trust the latest message. Do not mention this block unless asked.
The user's open goals:
- Ship the Kioko memory layer by December 15 (medium, 0%)
Relevant memories:
- Line 3 keeps faulting with error E-217.
- Found it: the encoder cable on line 3 was loose.
- The line 3 conveyor jammed twice this morning.
Procedures that may apply:
- Clear a conveyor jam on line 3: 1) Press stop; 2) Open the guard; 3) Remove the jammed bottle; 4) Reset the drive
Lessons from earlier attempts:
- For clearing conveyor jams: reset the drive before reopening the guard (1/1 worked). Lesson: always check the photo-eye alignment after a jam
How these memories connect:
- lesson "For clearing conveyor jams: reset the drive before reopenin…" was learned from event "The line 3 conveyor jammed twice this morning"
- event "Found it: the encoder cable on line 3 was loose" resolves event "Line 3 keeps faulting with error E-217"
- procedure "Clear a conveyor jam on line 3" applies to event "The line 3 conveyor jammed twice this morning"
... (11 more lines)