安装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存放外部数据，对话内容产生记忆。

1. 短期记忆
   就是最近 N 轮问答文本本身（ChatMemoryBuffer 里的 FIFO 队列）。
   用户每输入一个问题、模型每给出一次回答，都会被原封不动地压进队列；当总 token 数超过 SHORT_MEMORY_TOKENS 时，最老的整轮对话会被整体弹出——这就是即时遗忘。
2. 长期记忆
   由后台线程从“刚刚发生”的问答文本里二次加工产生：
   • 事实记忆：用非常简陋的「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))

解释

二、逐模块详细讲解

1. 模型配置
   与原来完全一致，只是将 OLLAMA_URL 等参数收归到顶部，方便一键修改。
2. 目录与常数
   • DATA_DIR：你的本地文档目录。
   • PERSIST_DIR：LlamaIndex 的索引持久化位置。
   • MEMORY_DIR：新增，用于存放「长期事实」和「向量存储」的本地文件。
   • SHORT_TOKENS / FACT_LIMIT / VECTOR_LIMIT：三处记忆硬阈值，直接决定“遗忘”节奏。
3. FactMemoryBlock
   用最简单的「三元组」模拟事实：
   • add()：同一 (主语, 谓词) 只保留最新宾语；超限后按「使用次数少 + 时间旧」淘汰。
   • retrieve()：支持模糊匹配主语，返回得分最高的 topk 条事实。
   • persist() / load()：纯文本 tsv 落盘，删除即“全脑遗忘”。
4. VectorMemoryBlock
   基于 SimpleVectorStore：
   • add()：任意文本 → 向量 → 入库。
   • retrieve()：先向量相似检索，再对每条结果按「时间衰减」重新打分，返回 topk。
   • _trim()：超限后整体重建向量库，保证只保留最近/最高分的 VECTOR_LIMIT 条。
5. HippocampusMemory（核心）
   把「短/长」记忆统一收口：
   • add_dialog()：每次对话后把整句写进短期 FIFO，并异步调用 _consolidate() 做「二次加工」→ 长期记忆。
   • add_documents()：新增函数，在构建索引后立即把整篇文档内容灌进长期记忆（既做向量也抽事实），实现「阅读即记忆」。
   • retrieve()：把三类记忆拼成一个字符串，直接塞给 LLM 当上下文。
   • periodic_persist()：定时落盘，防止进程崩溃丢失。
6. 索引管理
   build_index() 里多调用了一句 hippo.add_documents(documents)，于是每次目录变更、重新解析文档后，新内容会自动进入长期记忆；其余逻辑与原 demo 保持一致。
7. 后台线程 watcher
   仍负责「5 分钟扫一次目录」→ 有变化就重建索引 + 写记忆；扫完顺手 periodic_persist() 把记忆落盘。
8. 交互循环
   • 每次提问先把 hippo.retrieve(query) 拿出来当上下文。
   • 拿到回答后再把本轮 (问题, 回答) 写回记忆，形成闭环。

---

三、运行效果速写

复制

===  LlamaIndex + Ollama + 海马体记忆（q 退出）===
>>> 加载已有索引 …
问题: 什么是量子计算？
答: 量子计算是利用量子叠加与纠缠特性进行并行计算的新型计算范式……
（后台异步把该问答句做向量+事实抽取，写进长期记忆）

...5 min 后你在 data/ 放了新文件...
[index] 解析文档并构建索引 …
[index] 索引+记忆已更新   ← 新文档内容已自动成为长期记忆

此时再提问涉及新文档的主题，LLM 会同时拿到：

• 短期最近几轮对话
• 长期事实三元组
• 长期语义向量片段

实现「文档阅读 + 多轮对话」双层记忆的动态结合与自动遗忘。

VoxCPM 0.5B 可以多种方言 中文 英文

输入音频即可模拟发声

https://www.modelscope.cn/models/OpenBMB/VoxCPM-0.5B

# 1) 直接合成（单段文本）
voxcpm --text "Hello VoxCPM" --output out.wav

# 2) 声音克隆（参考音频 + 对应文本）
voxcpm --text "Hello" \
  --prompt-audio path/to/voice.wav \
  --prompt-text "reference transcript" \
  --output out.wav \
  --denoise

# 3) 批量处理（每行一段文本）
voxcpm --input examples/input.txt --output-dir outs
#（可选）批量 + 克隆
voxcpm --input examples/input.txt --output-dir outs \
  --prompt-audio path/to/voice.wav \
  --prompt-text "reference transcript" \
  --denoise

# 4) 推理参数（质量/速度）
voxcpm --text "..." --output out.wav \
  --cfg-value 2.0 --inference-timesteps 10 --normalize

# 5) 模型加载
# 优先使用本地路径
voxcpm --text "..." --output out.wav --model-path /path/to/VoxCPM_model_dir
# 或从 Hugging Face 自动下载/缓存
voxcpm --text "..." --output out.wav \
  --hf-model-id openbmb/VoxCPM-0.5B --cache-dir ~/.cache/huggingface --local-files-only

# 6) 降噪器控制
voxcpm --text "..." --output out.wav \
  --no-denoiser --zipenhancer-path iic/speech_zipenhancer_ans_multiloss_16k_base

# 7) 查看帮助
voxcpm --help
python -m voxcpm.cli --help

CMD中 使用 python -m voxcpm.cli 替换前面的voxcpm

克隆声音的范例

python -m voxcpm.cli --text "你好你在干什么啊" --prompt-audio a.wav --prompt-text "你好 现在是几点钟了 明天又是什么时候呢 大家 都在上班还是上学" --output out.wav --denoise

web测试页面

去：GitHub – OpenBMB/VoxCPM: VoxCPM: Tokenizer-Free TTS for Context-Aware Speech Generation and True-to-Life Voice Cloning

下载下来后，进入文件目录 执行python app.py即可运行

http://localhost:7860

qwen2.5-vl 7b 使用OCR 返回JSON信息 包含位置信息

为了防止幻觉：不使用提示词。

入参加上 response_format 来输出规定格式 及相对应的内容。

response_format={“type”: “json_object”}

$(“.chb:checked”).map(function(){return $(this).val()}).get();

$(".chb:checked").map(
    function(){
        return $(this).val()
    }
).get();

https://github.com/moeru-ai/airi

Project AIRI

模型驱动的灵魂容器，什么都能做一点的桌宠：让 Neuro-sama 这样的虚拟伴侣也成为我们世界中的一份子吧！

[加入 Discord] [试试看] [English] [日本語]

深受 Neuro-sama 启发

Warning

注意： 我们没有发行任何与本项目关联的加密货币或代币，请注意判断资讯并谨慎行事。

Note

我们有一个专门的组织 @proj-airi 用于所有从 Project AIRI 诞生的子项目，快来看看吧！

RAG（检索增强生成）、记忆系统、嵌入式数据库、图标、Live2D 实用工具等等！

你是否梦想过拥有一个赛博生命（网络老婆/老公、数字桌宠），或者能与你玩耍和交谈的 AI 伴侣？

借助现代大型语言模型的力量，像是 ChatGPT 和著名的 Claude 所能带来的，想要 LLM（大语言模型）和我们角色扮演、聊天已经超简单了，每个人都能上手。而像 Character.ai（又称 c.ai） 和 JanitorAI 这样的平台，以及本地应用如 SillyTavern（又称酒馆），已经是基于聊天或文字冒险游戏体验的相当不错的解决方案。

但是，如何赋予它们玩游戏的能力呢？让它们能看到你正在编写的代码？不仅能一边聊天一边玩游戏，也可以看视频，还能做很多其他事情？

你可能已经知道 Neuro-sama，她目前是最好的能够玩游戏、聊天并与你和参与者（在VTuber社区中）互动的 AI VTuber / 伴侣，有些人也称这种存在为”数字人”。可惜的是，她并不开源，当她从直播中下线后，你就无法与她互动了。

因此，这个项目 AIRI，在这里提供了另一种可能性：让你轻松拥有自己的数字生命、赛博生命，随时随地。

这个项目有什么特别的呢？

与其他 AI 和 LLM 驱动的 VTuber 开源项目不同，アイリ VTuber 从开始开发的第一天开始就支持多种 Web 技术，涵盖诸如 WebGPU、WebAudio、Web Workers、WebAssembly、WebSocket 等已经广泛应用或仍在大量实验的 API。

这意味着 アイリ VTuber 能够在现代浏览器和设备上运行，甚至能够在移动设备上运行（已经完成了 PWA 支持），这为我们（贡献者们）带来了更多的可能性，让我们得以更进一步构建和扩展 アイリ VTuber 的外部功能，而与此同时也不失配置的灵活性——可以有选择地在不同设备上启用会需要 TCP 连接或其他非 Web 技术的功能，例如连接到 Discord 的语音频道一起开黑，或是和朋友们一起玩 Minecraft（我的世界）、Factorio（异星工厂）。

Note

アイリ VTuber 仍处于早期开发阶段，我们欢迎优秀的开发者加入我们，一起将它变为现实。

即使不熟悉 Vue.js、TypeScript 和所需的其他开发工具也没关系，我们也欢迎艺术家、设计师、运营策划的加入，你甚至可以成为第一个用 アイリ VTuber 直播的博主。

如果你使用的是 React、 Svelte，甚至 Solid 也没关系，您可以自己创建一个子目录，添加您希望在 アイリ VTuber 中看到的功能，或者想实验的功能。

我们非常期待以下领域的朋友加入：

• Live2D 模型师
• VRM 模型师
• VRChat 模型设计师
• 计算机视觉（CV）
• 强化学习（RL）
• 语音识别
• 语音合成
• ONNX 推理运行时
• Transformers.js
• vLLM
• WebGPU
• Three.js
• WebXR (也可以看看我们在 @moeru-ai 组织下另外的这个项目)

如果你已经感兴趣了，为什么不来这里和大家打个招呼呢？Would like to join part of us to build AIRI?

当前进度

• 思维能力
  •  玩 Minecraft
  •  玩 Factorio
  •  在 Telegram 聊天
  •  在 Discord 聊天
  • 记忆
    •  纯浏览器内数据库支持（基于 DuckDB WASM 或者 sqlite）
    •  Alaya 记忆层（施工中）
  •  纯浏览器的本地推理（基于 WebGPU）
• 语音理解
  •  浏览器音频输入
  •  Discord 音频输入
  •  客户端语音识别
  •  客户端说话检测
• 语言能力
  •  ElevenLabs 语音合成
• 身体动作
  • VRM 支持
    •  控制 VRM 模型
  • VRM 模型动画
    •  自动眨眼
    •  自动看
    •  空闲眼睛移动
  • Live2D 支持
    •  控制 Live2D 模型
  • Live2D 模型动画
    •  自动眨眼
    •  自动看
    •  空闲眼睛移动

开发

有关开发此项目的具体教程，参见 CONTRIBUTING.md

pnpm i pnpm dev

网页版 (也就是 airi.moeru.ai 的版本)

pnpm dev:web

桌面版（也叫拓麻歌子，aka 电子宠物）

pnpm dev:tamagotchi

文档站

pnpm -F @proj-airi/docs dev

原生支持的 LLM API 服务来源列表（由 xsai 驱动）

•  OpenRouter
•  vLLM
•  SGLang
•  Ollama
•  Google Gemini
• OpenAI
  •  Azure OpenAI API
• Anthropic Claude
  •  AWS Claude
•  深度求索 DeepSeek
•  通义千问 Qwen
•  xAI
•  Groq
•  Mistral
•  Cloudflare Workers AI
•  Together.ai
•  Fireworks.ai
•  Novita
•  智谱
•  硅基流动
•  阶跃星辰
•  百川
•  Minimax
•  月之暗面
•  魔搭社区
•  Player2
•  腾讯混元
•  讯飞星火
•  火山引擎（豆包）

从这个项目诞生的子项目

• unspeech: 用于代理 /audio/transcriptions 和 /audio/speech 的代理服务器实现，类似 LiteLLM 但面向任何 ASR 和 TTS
• hfup: 帮助部署、打包到 HuggingFace Spaces 的工具集
• @proj-airi/drizzle-duckdb-wasm: DuckDB WASM 的 Drizzle ORM driver 驱动
• @proj-airi/duckdb-wasm: 易于使用的 @duckdb/duckdb-wasm 封装
• @proj-airi/lobe-icons: 为 lobe-icons 漂亮的 AI & LLM 图标制作的 Iconify JSON 封装，支持 Tailwind 和 UnoCSS
• AIRI Factorio: 让 AIRI 玩 Factorio
• Factorio RCON API: Factorio 无头服务器控制台的 RESTful API 封装
• autorio: Factorio 自动化库
• tstl-plugin-reload-factorio-mod: 开发时支持热重载 Factorio 模组
• 🥺 SAD: 自托管和浏览器运行 LLM 的文档和说明
• Velin: 用 Vue SFC 和 Markdown 文件来为 LLM 书写简单好用的提示词
• demodel: 轻松加速各种推理引擎和模型下载器拉/下载模型或数据集的速度
• inventory: 中心化模型目录和默认服务来源配置的公开 API 服务
• MCP Launcher: 易于使用的 MCP 启动器，适用于所有可能的 MCP Server，就像用于模型推理的 Ollama 一样！
• @proj-airi/elevenlabs: ElevenLabs API 的类型定义

https://viewscreen.githubusercontent.com/markdown/mermaid?docs_host=https%3A%2F%2Fdocs.github.com&color_mode=light#d44cc388-5696-4f6d-8a48-845520d580d2Loading

同类项目

开源项目

• kimjammer/Neuro: A recreation of Neuro-Sama originally created in 7 days.：非常完善的 Neuro-Sama 实现
• SugarcaneDefender/z-waif：以游戏、自主代理和提示词工程见长
• semperai/amica：适配 VRM, WebXR
• elizaOS/eliza：将 AI 智能体集成至各类系统和 API 中的一个软件工程实践
• ardha27/AI-Waifu-Vtuber：Twitch API 集成
• InsanityLabs/AIVTuber：优秀的 UI/UX 设计
• IRedDragonICY/vixevia
• t41372/Open-LLM-VTuber
• PeterH0323/Streamer-Sales

非开源项目

• https://clips.twitch.tv/WanderingCaringDeerDxCat-Qt55xtiGDSoNmDDr https://www.youtube.com/watch?v=8Giv5mupJNE
• https://clips.twitch.tv/TriangularAthleticBunnySoonerLater-SXpBk1dFso21VcWD

项目状态

鸣谢

• Reka UI: 的文档页面设计，新的着陆页基本上是在他们完成的工作至上发挥得来的，也感谢他们实现了大量的 UI 组件。（shadcn-vue 使用 Reka UI 作为 headless 组件，也请记得去看看！）
• pixiv/ChatVRM
• josephrocca/ChatVRM-js: A JS conversion/adaptation of parts of the ChatVRM (TypeScript) code for standalone use in OpenCharacters and elsewhere
• UI 和样式的设计受 Cookard，UNBEATABLE，以及 Sensei! I like you so much!，还有 Ayame by Mercedes Bazan 和 Wish by Mercedes Bazan 的作品启发
• mallorbc/whisper_mic
• xsai：实现了相当数量的包来与 LLM 和模型交互，像 Vercel AI SDK 但是更小

Star History