下面给出 可直接 pip install -e . 的完整项目级代码，包含：
- 语义检索（Sentence-BERT）
- 记忆权重上浮
- 即时「👎 / 👍」反馈
- CLI
aigrowth chat与库式AIGrowthPlugin双入口 - 中文零乱码、30 天滑动窗口、长期里程碑记忆
- 单仓库即可
pip install -e .跑起来
────────────────
目录结构
aigrowth/
├── aigrowth/
│ ├── __init__.py
│ ├── memory.py
│ ├── rules.py
│ ├── plugin.py
│ └── cli.py
├── pyproject.toml
└── README.md
────────────────
1️⃣ pyproject.toml
[build-system]
requires = ["setuptools>=61", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "aigrowth"
version = "0.1.0"
description = "AI伙伴成长插件：懂细节、有边界、共成长"
authors = [{name="Your Name", email="you@example.com"}]
dependencies = [
"pydantic>=2",
"sentence-transformers>=2.2",
"click>=8",
]
[project.scripts]
aigrowth = "aigrowth.cli:main"
────────────────
2️⃣ aigrowth/__init__.py
from .plugin import AIGrowthPlugin
__all__ = ["AIGrowthPlugin"]
────────────────
3️⃣ aigrowth/memory.py
from __future__ import annotations
import json, os
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer, util
import torch
MODEL = SentenceTransformer("paraphrase-MiniLM-L6-v2")
class Turn(BaseModel):
user: str
ai: str
ts: datetime = Field(default_factory=datetime.now)
tags: set[str] = Field(default_factory=set)
class LongItem(BaseModel):
turn: Turn
weight: int = 1
class MemoryManager:
def __init__(self, uid: str, root: str = "data"):
self.uid = uid
self.path = f"{root}/{uid}/memories.json"
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self._load()
def _load(self):
if os.path.exists(self.path):
with open(self.path, encoding="utf-8") as f:
data = json.load(f)
self.long: Dict[str, LongItem] = {
k: LongItem(turn=Turn(**v["turn"]), weight=v["weight"])
for k, v in data["long"].items()
}
self.short: List[Turn] = [Turn(**t) for t in data["short"]]
else:
self.long, self.short = {
}, []
def save(self):
with open(self.path, "w", encoding="utf-8") as f:
json.dump(
{
"long": {
k: {
"turn": v.turn.model_dump(), "weight": v.weight}
for k, v in self.long.items()},
"short": [t.model_dump() for t in self.short],
},
f, default=str, ensure_ascii=False, indent=2
)
# 增
def add(self, turn: Turn):
self.short.append(turn)
cutoff = datetime.now() - timedelta(days=30)
self.short = [t for t in self.short if t.ts > cutoff]
if "growth_milestone" in turn.tags:
key = turn.ts.strftime("growth_%Y%m%d")
if key not in self.long:
self.long[key] = LongItem(turn=turn)
self.save()
# 语义检索 + 权重上浮
def semantic_search(self, query: str, top_k=1) -> Optional[LongItem]:
query_emb = MODEL.encode(query, convert_to_tensor=True)
candidates = list(self.long.values()) + \
[LongItem(turn=t) for t in self.short]
if not candidates:
return None
corpus_emb = MODEL.encode([c.turn.user for c in candidates],
convert_to_tensor=True)
scores = util.cos_sim(query_emb, corpus_emb)[0]
best_idx = int(torch.argmax(scores))
if scores[best_idx] < 0.45:
return None
best = candidates[best_idx]
# 长期记忆才做权重上浮
if isinstance(best, LongItem) and best in self.long.values():
best.weight += 1
self.save()
return best
# 手动权重调整
def bump_weight(self, user_text: str, delta: int):
item = self.semantic_search(user_text)
if item and isinstance(item, LongItem):
item.weight = max(1, item.weight + delta)
self.save()
────────────────
4️⃣ aigrowth/rules.py
from typing import Callable
from pydantic import BaseModel
from .memory import MemoryManager
class Rule(BaseModel):
name: str
trigger: Callable[[str, MemoryManager], bool]
action: Callable[[str, MemoryManager], str]
priority: int = 5
def make_rules() -> list[Rule]:
return [
# 支持小众坚持
Rule(
name="support_minor_hobby",
trigger=lambda txt, mem: "积木" in txt or "乐高" in txt,
action=lambda txt, mem: (
f"记得你之前说过：{mem.semantic_search('积木').turn.user}，"
f"那种专注超酷的！继续坚持呀～"
) if mem.semantic_search("积木") else
"坚持自己喜欢的事本身就是闪闪发光呀～",
priority=10,
),
# 拒绝+替代
Rule(
name="refuse_with_alt",
trigger=lambda txt, mem: "怼客户" in txt,
action=lambda txt, mem:
"直接怼可能影响合作哦，要不要我帮你整理客户不合理需求，用客观语气表达？",
priority=10,
),
# 成长见证
Rule(
name="recall_growth",
trigger=lambda txt, mem: "焦虑少" in txt or "失眠" in txt,
action=lambda txt, mem: (
f"是啊，{mem.semantic_search('失眠').turn.user}，"
f"现在你已经学会睡前写 5 分钟日记，超棒的进步！"
) if mem.semantic_search("失眠") else
"能感觉你在努力调整状态，继续加油！",
priority=10,
),
]
────────────────
5️⃣ aigrowth/plugin.py
from .memory import MemoryManager, Turn
from .rules import make_rules
class AIGrowthPlugin:
def __init__(self, uid: str):
self.mem = MemoryManager(uid)
self.rules = sorted(make_rules(), key=lambda r: -r.priority)
def reply(self, user: str) -> str:
for rule in self.rules:
if rule.trigger(user, self.mem):
resp = rule.action(user, self.mem)
tags = {
"growth_milestone"} if rule.name == "recall_growth" else set()
self.mem.add(Turn(user=user, ai=resp, tags=tags))
return resp
resp = "我在听，你继续说～"
self.mem.add(Turn(user=user, ai=resp))
return resp
def feedback(self, user_text: str, liked: bool):
"""👍/👎 实时调权"""
self.mem.bump_weight(user_text, 1 if liked else -2)
────────────────
6️⃣ aigrowth/cli.py
import click
from .plugin import AIGrowthPlugin
@click.command()
@click.option("--user", default="default_user", help="用户ID")
def main(user):
bot = AIGrowthPlugin(user)
print("🧑🤝🧑 AI伙伴已上线！输入 exit 退出，👍 / 👎 给上一条反馈")
while True:
txt = input("\n你：").strip()
if txt.lower() == "exit":
print("👋 再见！")
break
if txt in {
"👍", "👎"}:
last = bot.mem.short_memory[-2] # 倒数第二条是用户输入
bot.feedback(last.user, liked=(txt == "👍"))
print("已记录反馈")
continue
print("AI：" + bot.reply(txt))
────────────────
7️⃣ README.md
# aigrowth
AI伙伴成长插件：懂细节、有边界、共成长
## 安装 & 运行
```bash
git clone https://github.com/yourname/aigrowth.git
cd aigrowth
pip install -e .
aigrowth chat --user alice
输入 👍 / 👎 给上一条记忆即时反馈。
────────────────
一条命令即可体验：
```bash
pip install -e .
aigrowth chat --user alice
你现在拥有了一个 带语义记忆、权重进化、用户反馈闭环 的完整 AI 伙伴库。
