在线咨询

专属客服在线解答，提供专业解决方案

工单支持

专业技术支持团队，随时响应服务需求

首页博客正文

通过函数调用 + SQLite：让语音智能体记住上次说的话

2026-07-27

开发者教程

「记住，周五下班前要发布教程」——用户说完这句，助手会答应得很好，然后在会话结束的那一刻忘得一干二净。

要让它真记住，得把几件事串起来：模型能调用你的业务函数，函数结果能落库，下次会话还能取回来。缺一件，「记住」就只是嘴上说说。

这篇教程走一遍完整的函数调用闭环：DeepSeek 自己判断要不要调 remember_note 或 list_notes，业务服务只执行白名单里的函数，结果写进 SQLite，再交回模型组织成一句能听的口语回答。

跑通之后你得到的是一个真能记事的助手——今天说的话，明天新开一个会话它还记得。

> 开始之前　HTTP 入口沿用《给语音智能体接入自定义 LLM：OpenAI 兼容网关》里的写法，先把那个网关跑通。

一. 架构与准备工作

声网 Custom LLM 请求
  → DeepSeek（tools + tool_choice:auto）
      ├─ 普通回答 → SSE
      └─ tool_calls → 校验名称/参数 → SQLite
                                      → tool result 回填 DeepSeek → SSE

环境变量：

DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SECRET
MESSAGE_DB_PATH=/data/messages.db

二. 工具定义与持久化

import json, os, sqlite3, time
from typing import Any

DB_PATH = os.getenv("MESSAGE_DB_PATH", "messages.db")
TOOLS = [
  {"type": "function", "function": {
    "name": "remember_note", "description": "保存用户要求记住的事项",
    "parameters": {"type": "object", "properties": {
      "text": {"type": "string", "description": "备忘内容"}}, "required": ["text"]}}},
  {"type": "function", "function": {
    "name": "list_notes", "description": "查询之前保存的备忘事项",
    "parameters": {"type": "object", "properties": {}}}},
]

def db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("CREATE TABLE IF NOT EXISTS notes "
                 "(id INTEGER PRIMARY KEY, text TEXT NOT NULL, created_at REAL NOT NULL)")
    return conn

def run_tool(name: str, arguments: dict[str, Any]) -> str:
    conn = db()
    try:
        if name == "remember_note":
            text = str(arguments.get("text", "")).strip()
            if not text: return "备忘内容为空，未保存。"
            conn.execute("INSERT INTO notes(text, created_at) VALUES (?, ?)",
                         (text, time.time()))
            conn.commit()
            return f"已保存：{text}"
        if name == "list_notes":
            rows = conn.execute(
                "SELECT text FROM notes ORDER BY created_at DESC LIMIT 10").fetchall()
            return "；".join(row[0] for row in rows) if rows else "暂无备忘。"
        return f"未授权的工具：{name}"
    except sqlite3.Error as exc:
        return f"工具执行失败：{type(exc).__name__}"
    finally:
        conn.close()

Custom LLM 的编排逻辑：

import httpx

async def orchestrate(payload):
    messages = [{"role": "system", "content":
        "要求记住事项时必须调用 remember_note；"
        "询问‘我让你记住了什么’、‘之前记了什么’或任何备忘历史时，"
        "即使当前上下文没有旧消息，也必须调用 list_notes，禁止猜测。"},
        *clean_messages(payload.get("messages", []))]
    first = {
        "model": "deepseek-v4-flash", "messages": messages,
        "tools": TOOLS, "tool_choice": "auto", "stream": False,
        "temperature": 0.2, "thinking": {"type": "disabled"},
    }
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            "https://api.deepseek.com/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['DEEPSEEK_API_KEY']}"},
            json=first)
        response.raise_for_status()
        assistant = response.json()["choices"][0]["message"]

    calls = assistant.get("tool_calls") or []
    if not calls:
        return assistant.get("content") or ""

    messages.append(assistant)
    for call in calls:
        try: arguments = json.loads(call["function"].get("arguments") or "{}")
        except json.JSONDecodeError: arguments = {}
        result = run_tool(call["function"]["name"], arguments)
        messages.append({"role": "tool", "tool_call_id": call["id"], "content": result})

    # 第二次请求将 messages 以 stream:true 发给 DeepSeek，
    # 再把上游 data: 块原样返回声网。
    return messages

这段编排的关键是「两次模型请求」：第一次模型返回 tool_calls，我们执行函数拿到结果，把结果作为 tool message 回填，第二次请求模型才把它说成人话。少了第二次请求，函数是执行了，但用户什么也听不到。

HTTP 入口和《给语音智能体接入自定义 LLM：OpenAI 兼容网关》一样：校验 Bearer 密钥、接收 /chat/completions、用 StreamingResponse 返回兼容 SSE。智能体那边继续用 CustomLLM(base_url=..., api_key=...)。

配套的 tool_service.py 里还实现了 safe_orchestrate()，在最外层兜住 DeepSeek 的超时和 401/429/5xx，转成标准 SSE 降级内容。上游报错的细节和凭证信息不会被播报给用户。这点在语音场景尤其重要，因为用户没法像看网页那样跳过错误信息。

三. 运行与验证

uvicorn tool_service:app --host 0.0.0.0 --port 9200

按顺序做三步：

说「记住：周五下班前发布教程」，智能体应该调用 remember_note 并确认。
结束当前会话，新建一个智能体会话。
问「之前记了什么？」，智能体应该调用 list_notes 并说出上一次记的内容。

第 2 步不能省，它才是这篇教程的重点：跨会话之后记忆还在，说明数据真的落库了，而不是留在上下文里。

上线前还要覆盖这几种情况：空参数、不在白名单里的工具名、非法 JSON、SQLite 写入失败、DeepSeek 返回 429/5xx。工具名、耗时、成功与否都可以记日志，但别记用户说的完整内容。

四. 故障排查

模型不调工具：检查 tool schema 是否合法、system prompt 有没有说清楚什么时候该调、以及模型本身支持不支持 tool calling。
工具执行了但没播报结果：确认 assistant 的 tool_calls 和每个 tool_call_id 对应的 tool message 都回填进了第二次模型请求。这是最常见的一种「静默失败」。
SQLite 只是最小示例。真接业务时必须按租户/用户隔离数据，并想清楚记忆保留多久、用户要删除时怎么删。

五. 下一步

函数会调了、结果也落库了，下一步是把记忆做安全：用户隔离和删除接口。

《语音智能体的跨会话记忆：持久化与隐私删除》

在声网，连接无限可能

想进一步了解「对话式 AI 与实时互动」？欢迎注册，开启探索之旅。

注册体验

本博客为技术交流与平台行业信息分享平台，内容仅供交流参考，文章内容不代表本公司立场和观点，亦不构成任何出版或销售行为。