综合实战：个人助理智能体

第二个综合项目更"像产品"：一个能用自然语言安排日程、查天气、记备忘的个人助理，还带上三件实战必需品——长期记忆（记住你上次聊过的偏好）、人工审批（删除备忘这类危险操作先问人）、会话持久化（重启后还能续聊）。核心不是"调一次模型"，而是让工具、记忆、审批、持久化四个部件协作。

功能与架构

CLI/对话层
   │  ① 召回记忆 ② 组装消息
   ▼
意图与工具循环（LLM 函数调用）
   ├── add_memo / list_memos     备忘（SQLite）
   ├── delete_memo  ★危险→人工审批
   ├── add_schedule/list_schedule 日程（SQLite）
   └── get_weather               天气（HTTP，见第 09 章）
   │  ③ 对话沉淀：偏好写入记忆
   ▼
SQLite：memos / schedules / chat_log / memory（偏好向量）

存储层：一张库四张表

# assistant_store.py
import json, os, sqlite3, time, zlib

DB = "assistant.db"

def conn():
    c = sqlite3.connect(DB)
    c.executescript("""
        CREATE TABLE IF NOT EXISTS memos(id INTEGER PRIMARY KEY AUTOINCREMENT,
            text TEXT, created TEXT);
        CREATE TABLE IF NOT EXISTS schedules(id INTEGER PRIMARY KEY AUTOINCREMENT,
            when_at TEXT, what TEXT);
        CREATE TABLE IF NOT EXISTS chat_log(session_id TEXT, role TEXT, content TEXT, ts TEXT);
        CREATE TABLE IF NOT EXISTS memory(key TEXT PRIMARY KEY, text TEXT,
            vector TEXT, updated_at TEXT);""")
    return c

def add_memo(text: str) -> str:      # 返回给模型看的可读结果
    c = conn(); c.execute("INSERT INTO memos(text, created) VALUES(?, ?)",
                          (text, time.strftime("%m-%d %H:%M"))); c.commit()
    n = c.execute("SELECT COUNT(*) FROM memos").fetchone()[0]
    return f"已记录备忘，当前共 {n} 条"

其余几个存储函数同样"一条 SQL、返回人话文本"（delete_memo 会先确认 id 存在，不存在返回"备忘不存在"）：

def add_schedule(when_at: str, what: str) -> str:
    c = conn(); c.execute("INSERT INTO schedules(when_at, what) VALUES(?,?)", (when_at, what)); c.commit()
    return f"已安排：{when_at} {what}"

def delete_memo(memo_id: int) -> str:
    c = conn(); cur = c.execute("DELETE FROM memos WHERE id=?", (memo_id,))
    c.commit()
    return "已删除该备忘" if cur.rowcount else "备忘不存在"

def list_memos() -> str:
    rows = conn().execute("SELECT id, text, created FROM memos ORDER BY id DESC LIMIT 10").fetchall()
    return "\n".join(f"#{i} {t}（{d}）" for i, t, d in rows) or "（暂无备忘）"

def list_schedule() -> str:
    rows = conn().execute("SELECT id, when_at, what FROM schedules ORDER BY id DESC LIMIT 10").fetchall()
    return "\n".join(f"#{i} {w} {t}" for i, w, t in rows) or "（暂无日程）"

工具只返回文本，模型才能自然地说给用户听——这是第 05 章"工具输出要为模型设计"的落地。

长期记忆：偏好写入与向量召回

沿用第 13 章"sqlite 存向量 + 余弦召回"的 B 路实现（离线 debug 向量，正式环境换 embedding 服务，接口不变）：

def _embed(text: str, dim: int = 128) -> list:
    v = [0.0] * dim                       # 教学伪向量：字符二元组哈希
    for i in range(len(text) - 1):
        v[zlib.crc32(text[i:i+2].encode()) % dim] += 1.0
    n = sum(x*x for x in v) ** 0.5 or 1.0
    return [x/n for x in v]

def remember(key: str, text: str) -> None:
    c = conn()
    c.execute("INSERT INTO memory(key,text,vector,updated_at) VALUES(?,?,?,?) "
              "ON CONFLICT(key) DO UPDATE SET text=excluded.text, vector=excluded.vector, updated_at=excluded.updated_at",
              (key, text, json.dumps(_embed(text)), time.strftime("%Y-%m-%d %H:%M:%S")))
    c.commit()

def recall(query: str, top_k: int = 2) -> list:
    qv = _embed(query); rows = conn().execute("SELECT text, vector FROM memory").fetchall()
    def cos(a, b):
        return sum(x*y for x, y in zip(a, b)) / ((sum(x*x for x in a)**.5 or 1) * (sum(y*y for y in b)**.5 or 1))
    return [t for t, _ in sorted(((t, cos(qv, json.loads(vec))) for t, vec in rows),
                                 key=lambda kv: kv[1], reverse=True)[:top_k]]

工具清单：给模型的 JSON Schema

TOOLS = [
    {"type": "function", "function": {"name": "add_memo", "description": "记录一条备忘",
     "parameters": {"type": "object", "properties": {"text": {"type": "string", "description": "备忘内容"}}, "required": ["text"]}}},
    {"type": "function", "function": {"name": "delete_memo", "description": "删除一条备忘（危险操作，执行前会向用户确认）",
     "parameters": {"type": "object", "properties": {"id": {"type": "integer"}}, "required": ["id"]}}},
    {"type": "function", "function": {"name": "add_schedule", "description": "新增日程",
     "parameters": {"type": "object", "properties": {"when_at": {"type": "string", "description": "时间如 明天 14:00"}, "what": {"type": "string"}}, "required": ["when_at", "what"]}}},
    {"type": "function", "function": {"name": "get_weather", "description": "查询城市天气",
     "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}},
    {"type": "function", "function": {"name": "list_memos", "description": "列出最近的备忘，用户问'我有哪些备忘'时调用",
     "parameters": {"type": "object", "properties": {}}}},
    {"type": "function", "function": {"name": "list_schedule", "description": "列出最近的日程，用户问'我有什么安排'时调用",
     "parameters": {"type": "object", "properties": {}}}},
]
DANGEROUS = {"delete_memo"}          # 需要人工审批的工具集

每个 description 都写清"何时用、参数含义、是否危险"——模型反复用错参数时，先改这里（呼应第 05、32 章结论）。

天气工具：一次真实 HTTP 调用

查天气不落库，每次实时请求免费接口 wttr.in（异常要兜底，不能让工具把异常抛给循环）：

def get_weather(city: str) -> str:
    """按城市查天气；网络失败返回兜底文案而非抛异常（第 09 章 HTTP 工具思路）。"""
    try:
        import requests
        r = requests.get(f"https://wttr.in/{city}?format=3&lang=zh", timeout=8)
        return r.text.strip() or f"{city} 暂无天气数据"
    except Exception:
        return "天气服务暂时不可用，请稍后再试"

工具执行循环 + 人工审批

危险操作执行前调用 approve_fn（CLI 里弹确认，Web 服务里返回审批请求交给前端，见第 19 章思路）：

def dispatch(name: str, args: dict, approve_fn) -> str:
    if name == "delete_memo" and not approve_fn(args):
        return "用户拒绝了删除操作，请向用户说明并停手。"
    return {"add_memo": lambda a: add_memo(a["text"]),
            "delete_memo": lambda a: delete_memo(a["id"]),
            "add_schedule": lambda a: add_schedule(a["when_at"], a["what"]),
            "get_weather": lambda a: get_weather(a["city"]),
            "list_memos": lambda a: list_memos(),
            "list_schedule": lambda a: list_schedule()}[name](args)

def ask_confirm(args: dict) -> bool:
    return input(f"⚠ 确认删除备忘 #{args.get('id')} 吗？(y/N) ").strip().lower() == "y"

删除类工具一律先确认再执行——模型只负责"提议"，人负责"拍板"。

对话循环：记忆 × 会话 × 工具

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("LLM_API_KEY"), base_url=os.getenv("LLM_BASE_URL"))
MODEL = os.getenv("LLM_MODEL", "deepseek-chat")
PREF_KEYWORDS = ("喜欢", "习惯", "不吃", "尽量", "每天")

def run_turn(session_id: str, user_text: str) -> str:
    c = conn()
    hits = recall(user_text)
    sys_extra = ("\n用户长期偏好（可参考）：" + "；".join(hits)) if hits else ""
    messages = [{"role": "system", "content": "你是个人助理，调用工具办事，回答简洁。" + sys_extra}]
    for role, content in c.execute(
            "SELECT role, content FROM chat_log WHERE session_id=? ORDER BY rowid DESC LIMIT 6",
            (session_id,)).fetchall()[::-1]:
        messages.append({"role": role, "content": content})
    messages.append({"role": "user", "content": user_text})

    while True:                                   # 工具循环（同第 06 章）
        resp = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
        msg = resp.choices[0].message
        if not msg.tool_calls:
            break
        messages.append(msg.model_dump(exclude_none=True))
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            result = dispatch(tc.function.name, args, ask_confirm)
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
    answer = msg.content or "（完成）"

    ts = time.strftime("%Y-%m-%d %H:%M:%S")       # 会话持久化 + 偏好沉淀
    c.executemany("INSERT INTO chat_log VALUES(?,?,?,?)",
                  [(session_id, "user", user_text, ts), (session_id, "assistant", answer, ts)])
    if any(k in user_text for k in PREF_KEYWORDS):
        remember("pref:" + user_text[:40], user_text)
    c.commit()
    return answer

每次提问前：查长期偏好 + 取最近 6 条历史补进消息；每次回答后：整段对话落库、发现偏好句式就写入记忆——重启进程、换天再聊，"上次说的都还记得"就来自这两张表。

对话轨迹示例（演示，实际措辞随模型）

我> 记个备忘：下周二下午健身，顺便把它排进日程
助手> 好的，两条都办好了；另外记住你习惯把健身放在下午。
      [内部] add_schedule(when_at="下周二 14:00", what="健身") → 已安排

我> 明天上海天气怎么样
助手> 稍等，我查一下。[内部] get_weather(city="上海") → 上海: ☀️ +26°C；明天上海晴，26℃，适合出门。

我> 帮我删掉"下周二健身"那条备忘
助手> [内部] list_memos → 先查到 #2 是"下周二下午健身"；决定调用 delete_memo(id=2) → 触发审批
      ⚠ 确认删除备忘 #2 吗？(y/N) n
      好的，已取消删除，不动你的数据。

我> 那日程里有健身吗？（次日重启进程再问，记忆仍在）
助手> [内部] recall("日程里有健身吗") → 命中"健身安排在下午"
      查到了：下周二下午有一条健身日程。

上线边界

• 多用户：所有表加 user_id 字段，所有查询带条件，别让 A 删到 B 的备忘；
• Web 化：把 ask_confirm 换成"返回审批请求体"，接第 19/28 章的 interrupt 与 /chat 模板即可；
• 审批要有超时与审计：谁批的、何时批的记一行日志，出问题可追溯；
• 长期记忆注意隐私：记忆内容等于用户画像，涉及敏感偏好要提供"查看/删除我的记忆"入口。 小结：个人助理 = SQLite 管好备忘/日程/会话 + 向量召回实现长期记忆 + 函数调用循环让模型编排工具 + 危险工具先过人工审批；四件事各自独立又通过"每次提问前召回、每次回答后沉淀"串成闭环——这也是生产级助理的最小骨架。