安装llama-index
pip install llama-index
安装ollama的对应模块
pip install llama-index-llms-ollama llama-index-embeddings-ollama
#安装模型
ollama pull embeddinggemma
ollama pull gemma:4b
#可以换模型
默认以当前的目录下的data 文件夹下文件（后期插入文件可以定时执行来刷新RAG）
创建测试代码 app.py
#!/usr/bin/env python3
# demo.py : LlamaIndex + Ollama 本地问答
import os
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
load_index_from_storage,
)
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding
# 1. 全局模型配置
OLLAMA_URL = "http://127.0.0.1:11434" # 默认端口
EMBED_MODEL = "embeddinggemma" # 向量模型
LLM_MODEL = "gemma3:4b" # 生成模型
Settings.embed_model = OllamaEmbedding(
model_name=EMBED_MODEL, base_url=OLLAMA_URL
)
Settings.llm = Ollama(
model=LLM_MODEL, base_url=OLLAMA_URL, request_timeout=360.0
)
# 2. 持久化目录（同一目录二次启动可直接加载索引，无需重新解析）
PERSIST_DIR = "./storage"
# 3. 加载或构建索引
if not os.path.exists(PERSIST_DIR):
print(">>> 首次运行，解析文档并构建索引 …")
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=PERSIST_DIR)
else:
print(">>> 发现已有索引，直接加载 …")
storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
index = load_index_from_storage(storage_context)
# 4. 创建查询引擎
query_engine = index.as_query_engine(similarity_top_k=3)
# 5. 简单交互
if __name__ == "__main__":
print("=== LlamaIndex + Ollama 本地问答 ===")
while True:
q = input("\n问题 (q 退出): ").strip()
if q.lower() == "q":
break
resp = query_engine.query(q)
print("答:", resp)
定时刷新 使用watchdog
pip install watchdog
定时刷新data目录
#!/usr/bin/env python3
# demo.py : LlamaIndex + Ollama 本地问答 + 定时自动刷新索引
import os, time, threading
from pathlib import Path
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
load_index_from_storage,
)
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.llama_index.embeddings.ollama import OllamaEmbedding
# -------------------- 1. 全局模型配置 --------------------
OLLAMA_URL = "http://127.0.0.1:11434"
EMBED_MODEL = "nomic-embed-text" # 可自行替换
LLM_MODEL = "llama3.1:8b" # 可自行替换
Settings.embed_model = OllamaEmbedding(model_name=EMBED_MODEL, base_url=OLLAMA_URL)
Settings.llm = Ollama(model=LLM_MODEL, base_url=OLLAMA_URL, request_timeout=360.0)
# -------------------- 2. 参数 --------------------
DATA_DIR = Path("data")
PERSIST_DIR = Path("./storage")
RESCAN_INTERVAL = 5 * 60 # 秒，5 分钟扫一次
# -------------------- 3. 索引管理 --------------------
def build_index():
"""重新解析文档并持久化索引"""
print("[index] 正在重新构建索引 …")
documents = SimpleDirectoryReader(str(DATA_DIR)).load_data()
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=str(PERSIST_DIR))
print("[index] 索引已更新完成")
return index
def load_or_build():
"""启动时：有缓存就加载，没有就新建"""
if PERSIST_DIR.exists():
print(">>> 发现已有索引，直接加载 …")
storage_context = StorageContext.from_defaults(persist_dir=str(PERSIST_DIR))
return load_index_from_storage(storage_context)
else:
print(">>> 首次运行，解析文档并构建索引 …")
return build_index()
# -------------------- 4. 定时刷新线程 --------------------
def md5_dir():
"""简易指纹：拼接所有文件「路径+修改时间」"""
return "|".join(
f"{p.relative_to(DATA_DIR)}:{p.stat().st_mtime}"
for p in sorted(DATA_DIR.rglob("*"))
if p.is_file()
)
def watcher():
"""后台线程：循环扫描目录，有变化就重建"""
last_md5 = md5_dir()
while True:
time.sleep(RESCAN_INTERVAL)
new_md5 = md5_dir()
if new_md5 != last_md5:
last_md5 = new_md5
global index, query_engine
index = build_index()
query_engine = index.as_query_engine(similarity_top_k=3)
else:
print("[index] 目录无变化，跳过刷新")
# -------------------- 5. 启动 --------------------
index = load_or_build()
query_engine = index.as_query_engine(similarity_top_k=3)
threading.Thread(target=watcher, daemon=True).start()
# -------------------- 6. 交互 --------------------
if __name__ == "__main__":
print("=== LlamaIndex + Ollama 本地问答 (定时刷新已开启) ===")
while True:
q = input("\n问题 (q 退出): ").strip()
if q.lower() == "q":
break
resp = query_engine.query(q)
print("答:", resp)
模拟人类记忆 海马体
其中 data存放外部数据，对话内容产生记忆。
- 短期记忆
就是最近 N 轮问答文本本身（ChatMemoryBuffer 里的 FIFO 队列）。
用户每输入一个问题、模型每给出一次回答，都会被原封不动地压进队列；当总 token 数超过 SHORT_MEMORY_TOKENS 时，最老的整轮对话会被整体弹出——这就是即时遗忘。
- 长期记忆
由后台线程从“刚刚发生”的问答文本里二次加工产生：
- 事实记忆：用非常简陋的「A is B」正则把句子拆成 (主语, 谓词, 宾语) 三元组，再写进
FactMemoryBlock；容量满或长期未用即被淘汰。
- 语义向量：把整句/片段直接做成向量入库，后续用相似度召回；分数会随时间衰减，低于阈值或总量超限即被删除。
也就是说，没有对话就没有任何记忆可存；文档（data/目录）只提供“外部知识”，不会被自动写回记忆系统。
如果你想让模型把阅读到的文档内容也记到长期记忆里，只需要在 build_index() 之后把 documents 遍历一遍，手动调用：
for doc in documents:
hippo.vector.add(doc.text) # 语义记忆
# 或抽实体后 hippo.fact.add(…) # 事实记忆
#!/usr/bin/env python3
# demo_memory.py : LlamaIndex + Ollama + 海马体式「记忆–遗忘」
import os
import time
import threading
from collections import deque
from datetime import datetime
from pathlib import Path
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
load_index_from_storage,
)
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.schema import TextNode
from llama_index.core.vector_stores import SimpleVectorStore
# -------------------- 1. 全局模型 --------------------
OLLAMA_URL = "http://127.0.0.1:11434"
EMBED_MODEL = "nomic-embed-text"
LLM_MODEL = "llama3.1:8b"
Settings.embed_model = OllamaEmbedding(model_name=EMBED_MODEL, base_url=OLLAMA_URL)
Settings.llm = Ollama(model=LLM_MODEL, base_url=OLLAMA_URL, request_timeout=360.0)
# -------------------- 2. 参数 --------------------
DATA_DIR = Path("data")
PERSIST_DIR = Path("./storage")
MEMORY_DIR = PERSIST_DIR / "memory" # 长期记忆持久化
RESCAN_INTERVAL = 5 * 60
SHORT_MEMORY_TOKENS = 2000 # 短期记忆 token 上限
FACT_LIMIT = 200 # 长期事实上限
VECTOR_LIMIT = 1000 # 语义向量上限
DECAY_HALF_DAY = 0.9 # 每 12h 衰减系数
# -------------------- 3. 轻量级 MemoryBlock --------------------
class FactMemoryBlock:
"""结构化事实：【subject】【predicate】【object】+ 时间戳 + 使用次数"""
def __init__(self, limit=FACT_LIMIT):
self.limit = limit
self.facts = [] # [(sub, pred, obj, ts, count), ...]
self.load()
def add(self, sub, pred, obj):
now = datetime.utcnow()
# 简单合并：同一 (sub,pred) 只保留最新 obj
self.facts = [(s, p, o, ts, c) for (s, p, o, ts, c) in self.facts if not (s == sub and p == pred)]
self.facts.append((sub, pred, obj, now, 1))
if len(self.facts) > self.limit:
# 淘汰最久未用（count 最小且最旧）
self.facts.sort(key=lambda x: (x[4], x[3]))
self.facts = self.facts[-self.limit:]
def retrieve(self, query_sub, topk=5):
"""朴素字符串匹配 + 使用次数加权"""
hits = []
for sub, pred, obj, ts, c in self.facts:
score = c
if query_sub.lower() in sub.lower():
score += 5
hits.append((score, sub, pred, obj))
hits.sort(reverse=True)
return hits[:topk]
def persist(self):
MEMORY_DIR.mkdir(exist_ok=True)
with open(MEMORY_DIR / "facts.txt", "w", encoding="utf8") as f:
for sub, pred, obj, ts, c in self.facts:
f.write(f"{sub}\t{pred}\t{obj}\t{ts.isoformat()}\t{c}\n")
def load(self):
if (MEMORY_DIR / "facts.txt").exists():
with open(MEMORY_DIR / "facts.txt", encoding="utf8") as f:
for line in f:
sub, pred, obj, ts_str, c = line.strip().split("\t")
self.facts.append((sub, pred, obj, datetime.fromisoformat(ts_str), int(c)))
class VectorMemoryBlock:
"""语义向量：用 SimpleVectorStore 存节点，附加时间戳 & 分数"""
def __init__(self, limit=VECTOR_LIMIT, decay=DECAY_HALF_DAY):
self.limit = limit
self.decay = decay
self.store = SimpleVectorStore()
self.nodes = [] # 本地缓存 (node, ts, score)
if (MEMORY_DIR / "vector_store.json").exists():
self.store = SimpleVectorStore.from_persist_path(str(MEMORY_DIR / "vector_store.json"))
# 简版：只加载最近 limit 条
self._trim()
def add(self, text):
now = datetime.utcnow()
node = TextNode(text=text)
self.nodes.append((node, now, 1.0))
self.store.add([node])
self._trim()
def retrieve(self, query_str, top_k=3):
"""向量检索 + 时间衰减分数重排"""
emb = Settings.embed_model.get_text_embedding(query_str)
hits = self.store.query(embedding=emb, similarity_top_k=top_k * 2)
res = []
for n, ts, score in self.nodes:
hours = (datetime.utcnow() - ts).total_seconds() / 3600
decayed = score * (self.decay ** (hours / 12))
for hit in hits:
if hit.node.node_id == n.node_id:
res.append((decayed * hit.score, n.text))
break
res.sort(reverse=True)
return [txt for _, txt in res[:top_k]]
def _trim(self):
if len(self.nodes) > self.limit:
# 按分数+时间排序淘汰
self.nodes.sort(key=lambda x: (x[2], x[1]))
drop = self.nodes[:len(self.nodes) - self.limit]
self.nodes = self.nodes[-self.limit:]
# 同步从 store 删除（简单重建）
self.store = SimpleVectorStore()
self.store.add([n for n, _, _ in self.nodes])
def persist(self):
MEMORY_DIR.mkdir(exist_ok=True)
self.store.persist(str(MEMORY_DIR / "vector_store.json"))
# -------------------- 4. 记忆管理器 --------------------
class HippocampusMemory:
def __init__(self):
self.short = ChatMemoryBuffer(token_limit=SHORT_MEMORY_TOKENS)
self.fact = FactMemoryBlock()
self.vector = VectorMemoryBlock()
def add_interaction(self, user, assistant):
# 短期记忆
self.short.put(f"User: {user}\nAssistant: {assistant}")
# 异步长期巩固（简单抽实体）
threading.Thread(target self._consolidate, args=(user + " " + assistant,), daemon=True).start()
def _consolidate(self, text):
"""简易实体→事实抽取 & 语义片段入库"""
# 这里用 LLM 抽实体/摘要，为演示直接按逗号切
for sent in text.split("."):
sent = sent.strip()
if len(sent) < 10:
continue
# 向量记忆
self.vector.add(sent)
# 事实记忆：简单「A 是 B」模式
if " is " in sent:
sub, obj = sent.split(" is ", 1)
self.fact.add(sub.strip(), "is", obj.strip())
def retrieve_context(self, query):
# 短期
short_hist = self.short.get()
# 长期
facts = self.fact.retrieve(query)
vectors = self.vector.retrieve(query)
long_part = "\n".join([f"{s} {p} {o}" for _, s, p, o in facts]) + "\n" + "\n".join(vectors)
return f"短期记忆:\n{short_hist}\n\n长期记忆:\n{long_part}"
def periodic_forget(self):
"""定时遗忘：长期记忆自己维护容量 & 衰减"""
self.fact.persist()
self.vector.persist()
hippo = HippocampusMemory()
# -------------------- 5. 索引管理（同你原来） --------------------
def build_index():
print("[index] 正在重新构建索引 …")
documents = SimpleDirectoryReader(str(DATA_DIR)).load_data()
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=str(PERSIST_DIR))
return index
def load_or_build():
if PERSIST_DIR.exists():
storage_context = StorageContext.from_defaults(persist_dir=str(PERSIST_DIR))
return load_index_from_storage(storage_context)
else:
return build_index()
# -------------------- 6. 后台线程 --------------------
def md5_dir():
return "|".join(f"{p.relative_to(DATA_DIR)}:{p.stat().st_mtime}" for p in sorted(DATA_DIR.rglob("*")) if p.is_file())
def watcher():
last_md5 = md5_dir()
while True:
time.sleep(RESCAN_INTERVAL)
new_md5 = md5_dir()
if new_md5 != last_md5:
last_md5 = new_md5
global index
index = build_index()
hippo.periodic_forget()
# -------------------- 7. 启动 --------------------
index = load_or_build()
threading.Thread(target=watcher, daemon=True).start()
# -------------------- 8. 交互 --------------------
if __name__ == "__main__":
print("=== LlamaIndex + Ollama + 海马体记忆 (q 退出) ===")
while True:
q = input("\n问题: ").strip()
if q.lower() == "q":
break
# 把记忆作为上下文拼到 prompt
context = hippo.retrieve_context(q)
prompt = f"以下背景知识供参考：\n{context}\n\n用户问题：{q}"
resp = index.as_query_engine(similarity_top_k=3).query(prompt)
print("答:", resp)
hippo.add_interaction(q, str(resp))
结合文档变成长期记忆
#!/usr/bin/env python3
# hippo_bot.py : LlamaIndex + Ollama + 海马体记忆（对话产生 + 文档产生）
import os
import time
import threading
from collections import deque
from datetime import datetime
from pathlib import Path
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
load_index_from_storage,
Document,
)
from llama_index.llms.ollama import Ollama
from llama_index.embeddings.ollama import OllamaEmbedding
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.schema import TextNode
from llama_index.core.vector_stores import SimpleVectorStore
# -------------------- 1. 模型配置 --------------------
OLLAMA_URL = "http://127.0.0.1:11434"
EMBED_MODEL = "nomic-embed-text"
LLM_MODEL = "llama3.1:8b"
Settings.embed_model = OllamaEmbedding(model_name=EMBED_MODEL, base_url=OLLAMA_URL)
Settings.llm = Ollama(model=LLM_MODEL, base_url=OLLAMA_URL, request_timeout=360.0)
# -------------------- 2. 目录/常数 --------------------
DATA_DIR = Path("data")
PERSIST_DIR = Path("./storage")
MEMORY_DIR = PERSIST_DIR / "memory"
RESCAN_INTERVAL = 5 * 60 # 5 分钟扫一次
SHORT_TOKENS = 2000 # 短期记忆 token 上限
FACT_LIMIT = 200 # 事实上限
VECTOR_LIMIT = 1000 # 语义片上限
DECAY = 0.9 # 每 12h 衰减系数
MEMORY_DIR.mkdir(parents=True, exist_ok=True)
# -------------------- 3. 长期记忆块（同上一版） ----------
class FactMemoryBlock:
def __init__(self, limit=FACT_LIMIT):
self.limit = limit
self.facts = [] # [(sub, pred, obj, ts, count), ...]
self.load()
def add(self, sub, pred, obj):
now = datetime.utcnow()
self.facts = [(s, p, o, ts, c) for (s, p, o, ts, c) in self.facts if not (s == sub and p == pred)]
self.facts.append((sub.strip(), pred.strip(), obj.strip(), now, 1))
if len(self.facts) > self.limit:
self.facts.sort(key=lambda x: (x[4], x[3]))
self.facts = self.facts[-self.limit:]
def retrieve(self, query_sub, topk=5):
hits = []
for sub, pred, obj, ts, c in self.facts:
score = c
if query_sub.lower() in sub.lower():
score += 5
hits.append((score, sub, pred, obj))
hits.sort(reverse=True)
return hits[:topk]
def persist(self):
with open(MEMORY_DIR / "facts.txt", "w", encoding="utf8") as f:
for sub, pred, obj, ts, c in self.facts:
f.write(f"{sub}\t{pred}\t{obj}\t{ts.isoformat()}\t{c}\n")
def load(self):
p = MEMORY_DIR / "facts.txt"
if p.exists():
with p.open(encoding="utf8") as f:
for line in f:
sub, pred, obj, ts_str, c = line.strip().split("\t")
self.facts.append((sub, pred, obj, datetime.fromisoformat(ts_str), int(c)))
class VectorMemoryBlock:
def __init__(self, limit=VECTOR_LIMIT, decay=DECAY):
self.limit = limit
self.decay = decay
self.store = SimpleVectorStore()
self.nodes = [] # [(node, ts, score), ...]
if (MEMORY_DIR / "vector_store.json").exists():
self.store = SimpleVectorStore.from_persist_path(str(MEMORY_DIR / "vector_store.json"))
self._trim()
def add(self, text):
now = datetime.utcnow()
node = TextNode(text=text)
self.nodes.append((node, now, 1.0))
self.store.add([node])
self._trim()
def retrieve(self, query_str, top_k=3):
emb = Settings.embed_model.get_text_embedding(query_str)
hits = self.store.query(embedding=emb, similarity_top_k=top_k * 2)
res = []
for n, ts, score in self.nodes:
hours = (datetime.utcnow() - ts).total_seconds() / 3600
decayed = score * (self.decay ** (hours / 12))
for hit in hits:
if hit.node.node_id == n.node_id:
res.append((decayed * hit.score, n.text))
break
res.sort(reverse=True)
return [txt for _, txt in res[:top_k]]
def _trim(self):
if len(self.nodes) > self.limit:
self.nodes.sort(key=lambda x: (x[2], x[1]))
drop = self.nodes[:len(self.nodes) - self.limit]
self.nodes = self.nodes[-self.limit:]
self.store = SimpleVectorStore()
self.store.add([n for n, _, _ in self.nodes])
def persist(self):
self.store.persist(str(MEMORY_DIR / "vector_store.json"))
# -------------------- 4. 海马体记忆管理器 --------------------
class HippocampusMemory:
def __init__(self):
self.short = ChatMemoryBuffer(token_limit=SHORT_TOKENS)
self.fact = FactMemoryBlock()
self.vector = VectorMemoryBlock()
# 1. 对话产生的记忆
def add_dialog(self, user: str, assistant: str):
self.short.put(f"User: {user}\nAssistant: {assistant}")
# 异步巩固
threading.Thread(target=self._consolidate, args=(user + " " + assistant,), daemon=True).start()
# 2. 文档产生的记忆
def add_documents(self, docs: list[Document]):
for doc in docs:
self.vector.add(doc.text)
# 简单抽「A is B」
for sent in doc.text.split("."):
if " is " in sent:
sub, obj = sent.split(" is ", 1)
self.fact.add(sub.strip(), "is", obj.strip())
def _consolidate(self, text: str):
for sent in text.split("."):
sent = sent.strip()
if len(sent) < 10:
continue
self.vector.add(sent)
if " is " in sent:
sub, obj = sent.split(" is ", 1)
self.fact.add(sub.strip(), "is", obj.strip())
def retrieve(self, query: str) -> str:
short = self.short.get()
facts = self.fact.retrieve(query)
vectors = self.vector.retrieve(query)
long = "\n".join([f"{s} {p} {o}" for _, s, p, o in facts]) + "\n" + "\n".join(vectors)
return f"短期记忆:\n{short}\n\n长期记忆:\n{long}"
def periodic_persist(self):
self.fact.persist()
self.vector.persist()
hippo = HippocampusMemory()
# -------------------- 5. 索引管理 --------------------
def build_index():
print("[index] 解析文档并构建索引 …")
documents = SimpleDirectoryReader(str(DATA_DIR)).load_data()
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=str(PERSIST_DIR))
# ❗新增：把文档内容也写进长期记忆
hippo.add_documents(documents)
print("[index] 索引+记忆已更新")
return index
def load_or_build():
if PERSIST_DIR.exists():
print(">>> 加载已有索引 …")
storage_context = StorageContext.from_defaults(persist_dir=str(PERSIST_DIR))
return load_index_from_storage(storage_context)
else:
return build_index()
# -------------------- 6. 后台定时线程 --------------------
def md5_dir():
return "|".join(f"{p.relative_to(DATA_DIR)}:{p.stat().st_mtime}" for p in sorted(DATA_DIR.rglob("*")) if p.is_file())
def watcher():
last_md5 = md5_dir()
while True:
time.sleep(RESCAN_INTERVAL)
new_md5 = md5_dir()
if new_md5 != last_md5:
last_md5 = new_md5
global index
index = build_index()
else:
print("[index] 无变化，跳过刷新")
hippo.periodic_persist()
index = load_or_build()
threading.Thread(target=watcher, daemon=True).start()
# -------------------- 7. 交互 --------------------------
if __name__ == "__main__":
print("=== LlamaIndex + Ollama + 海马体记忆（q 退出）===")
while True:
q = input("\n问题: ").strip()
if q.lower() == "q":
break
context = hippo.retrieve(q)
prompt = f"以下背景知识供参考：\n{context}\n\n用户问题：{q}"
resp = index.as_query_engine(similarity_top_k=3).query(prompt)
print("答:", resp)
hippo.add_dialog(q, str(resp))
解释
二、逐模块详细讲解
- 模型配置
与原来完全一致，只是将 OLLAMA_URL 等参数收归到顶部，方便一键修改。
- 目录与常数
DATA_DIR：你的本地文档目录。PERSIST_DIR：LlamaIndex 的索引持久化位置。MEMORY_DIR：新增，用于存放「长期事实」和「向量存储」的本地文件。SHORT_TOKENS / FACT_LIMIT / VECTOR_LIMIT：三处记忆硬阈值，直接决定“遗忘”节奏。
- FactMemoryBlock
用最简单的「三元组」模拟事实：
add()：同一 (主语, 谓词) 只保留最新宾语；超限后按「使用次数少 + 时间旧」淘汰。retrieve()：支持模糊匹配主语，返回得分最高的 topk 条事实。persist() / load()：纯文本 tsv 落盘，删除即“全脑遗忘”。
- VectorMemoryBlock
基于 SimpleVectorStore：
add()：任意文本 → 向量 → 入库。retrieve()：先向量相似检索，再对每条结果按「时间衰减」重新打分，返回 topk。_trim()：超限后整体重建向量库，保证只保留最近/最高分的 VECTOR_LIMIT 条。
- HippocampusMemory（核心）
把「短/长」记忆统一收口：
add_dialog()：每次对话后把整句写进短期 FIFO，并异步调用 _consolidate() 做「二次加工」→ 长期记忆。add_documents()：新增函数，在构建索引后立即把整篇文档内容灌进长期记忆（既做向量也抽事实），实现「阅读即记忆」。retrieve()：把三类记忆拼成一个字符串，直接塞给 LLM 当上下文。periodic_persist()：定时落盘，防止进程崩溃丢失。
- 索引管理
build_index() 里多调用了一句 hippo.add_documents(documents)，于是每次目录变更、重新解析文档后，新内容会自动进入长期记忆；其余逻辑与原 demo 保持一致。
- 后台线程 watcher
仍负责「5 分钟扫一次目录」→ 有变化就重建索引 + 写记忆；扫完顺手 periodic_persist() 把记忆落盘。
- 交互循环
- 每次提问先把
hippo.retrieve(query) 拿出来当上下文。
- 拿到回答后再把本轮
(问题, 回答) 写回记忆，形成闭环。
三、运行效果速写
复制
=== LlamaIndex + Ollama + 海马体记忆（q 退出）===
>>> 加载已有索引 …
问题: 什么是量子计算？
答: 量子计算是利用量子叠加与纠缠特性进行并行计算的新型计算范式……
（后台异步把该问答句做向量+事实抽取，写进长期记忆）
...5 min 后你在 data/ 放了新文件...
[index] 解析文档并构建索引 …
[index] 索引+记忆已更新 ← 新文档内容已自动成为长期记忆
此时再提问涉及新文档的主题，LLM 会同时拿到：
- 短期最近几轮对话
- 长期事实三元组
- 长期语义向量片段
实现「文档阅读 + 多轮对话」双层记忆的动态结合与自动遗忘。