ylhan
98bbba93d0
按 P1-Pipecat Flows Agent化与记忆改造交接.md 的 Phase 4 完成。 新增 p1/phase4_acceptance.py：跑真实服务端验 7 条，每条之间查数据库。 实测结果（2026-09-16） 1. "记住我不吃香菜" → confirmed 落库 ✅ 2. **新连接** "我不吃什么？" → "你不吃香菜。" ✅ 连接开始时加载档案生效 3. "我今天吃了香菜" → 不新增 confirmed，模型还追问了是否口味变了 ✅ 4. "其实我现在吃香菜了" → 同键仍只一条 confirmed，值已更新 ✅ 5. "忘掉我的饮食偏好" → 已删除 ✅ 6/7. 访客与跨用户隔离 → 由单测覆盖（客户端不支持切身份） 实现：会话开始时加载稳定档案 - load_stable_memories()：取已确认、未过期记忆（身份未确认返回空） - _conversation_node 改用 build_system_prompt，与 legacy 共用同一套 人格与记忆注入格式；记忆进**同一条** system 消息，仍不用 role_message 修掉一个真实缺陷：Flows 路径误用了语音的 max_tokens - 症状：一句"忘掉我的饮食偏好"，服务端日志 "Failed to parse function call arguments: {"query": "饮食"（JSON 被截断）， 回复为空、记忆没删掉，且**没有任何报错** - 根因：Flows 路径沿用了语音的 max_tokens=200（为"回复要短"而设）。 原始 API 对照实测：max_tokens=200 时模型 finish=stop、根本不调工具； 600 时正常产生 tool_calls 且参数 JSON 完整 - 修法：新增 AgentConfig.max_tokens（AI_AGENT_MAX_TOKENS，默认 600）， Flows 路径用它，语音路径仍用 200 新增 client/test_client.py 能力（为验收所需） - RTVIReplies：收服务端 RTVI 文本消息（bot-output / bot-transcription） - 打印 BOT_REPLY 供验收脚本断言 回归：test_agent_tools 45/45、test_flows_runtime 24/24、 test_memory 30/30、test_identity 19/19 服务器已改回 AI_AGENT_RUNTIME=legacy（改默认是 Phase 7 的事）
271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""配置加载。
|
||
|
||
原则（来自方案 §10）：
|
||
- 密钥保存在本机配置，不写日志，不进版本管理
|
||
- 设备标识用稳定名字，不用会变的序号
|
||
- 依赖版本固定，板卡/固件/通道/引脚都记录
|
||
|
||
密钥文件是 ~/.config/ai_assistant/env（权限 600），格式是 shell 变量。
|
||
本模块只读它，不写。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
ENV_FILE = Path.home() / ".config" / "ai_assistant" / "env"
|
||
|
||
# 本项目根目录（app/ 的上一级）
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
|
||
_lock = threading.Lock()
|
||
_loaded = False
|
||
|
||
|
||
def load_env(path: Path | None = None) -> dict[str, str]:
|
||
"""读取密钥文件到 os.environ（已存在的变量不覆盖）。
|
||
|
||
只做极简解析：KEY=VALUE，忽略空行和 # 注释。
|
||
不用 os.system/source，避免把 shell 引进来。
|
||
"""
|
||
global _loaded
|
||
p = path or ENV_FILE
|
||
with _lock:
|
||
if _loaded:
|
||
return dict(os.environ)
|
||
if p.is_file():
|
||
for line in p.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
k, v = k.strip(), v.strip().strip('"').strip("'")
|
||
os.environ.setdefault(k, v)
|
||
_loaded = True
|
||
return dict(os.environ)
|
||
|
||
|
||
def _env(key: str, default: str = "") -> str:
|
||
load_env()
|
||
return os.environ.get(key, default)
|
||
|
||
|
||
def _env_int(key: str, default: int) -> int:
|
||
try:
|
||
return int(_env(key, str(default)))
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def _env_float(key: str, default: float) -> float:
|
||
try:
|
||
return float(_env(key, str(default)))
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def _env_bool(key: str, default: bool) -> bool:
|
||
value = _env(key, "true" if default else "false").strip().lower()
|
||
return value in {"1", "true", "yes", "on"}
|
||
|
||
|
||
# ---------------------------------------------------------------- LLM
|
||
|
||
@dataclass
|
||
class LLMConfig:
|
||
"""一个 LLM 后端的配置。
|
||
|
||
thinking 的处理方式各后端不同，所以由后端自己解释这个开关：
|
||
- 本机 vLLM： chat_template_kwargs={"enable_thinking": false}
|
||
- DeepSeek： thinking={"type": "disabled"}
|
||
两者都实测有效，见 P0-本地LLM验证报告.md 与 P0-DeepSeek验证报告.md。
|
||
"""
|
||
name: str
|
||
base_url: str
|
||
api_key: str
|
||
model: str
|
||
thinking: bool = False # 语音场景默认关思考
|
||
max_tokens: int = 200 # 语音回复要短，方案要求"默认简短自然回复"
|
||
timeout_s: float = 30.0
|
||
first_token_timeout_s: float = 8.0
|
||
|
||
@property
|
||
def enabled(self) -> bool:
|
||
return bool(self.base_url and self.api_key and self.model)
|
||
|
||
|
||
def local_llm_config() -> LLMConfig:
|
||
"""本机/内网 LLM（vLLM 等）。
|
||
|
||
默认走回环：服务与助手同机时 127.0.0.1 最省一跳。
|
||
部署到别的机器时必须用 LOCAL_LLM_BASE_URL 显式覆盖——
|
||
**不要把具体的内网地址写进默认值**，那会把网络拓扑带进版本历史，
|
||
而且换环境就得改代码。
|
||
"""
|
||
return LLMConfig(
|
||
name="local",
|
||
base_url=_env("LOCAL_LLM_BASE_URL", "http://127.0.0.1:18020/v1"),
|
||
api_key=_env("LOCAL_LLM_API_KEY"),
|
||
model=_env("LOCAL_LLM_MODEL", "qwen3.8-27b"),
|
||
)
|
||
|
||
|
||
def deepseek_config() -> LLMConfig:
|
||
return LLMConfig(
|
||
name="deepseek",
|
||
base_url=_env("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1"),
|
||
api_key=_env("DEEPSEEK_API_KEY"),
|
||
model=_env("DEEPSEEK_MODEL", "deepseek-flash"),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------- 上下文
|
||
|
||
@dataclass
|
||
class ContextConfig:
|
||
"""上下文预算。
|
||
|
||
实测：真实语音 prompt（人设+记忆+3轮对话）约 406 tokens，
|
||
本机 LLM 首 token 367ms。上下文越长，首 token 越慢——直接伤"回应及时"。
|
||
所以这里设的是硬上限，超出就丢最旧的轮次，不是软建议。
|
||
"""
|
||
max_prompt_tokens: int = 4096 # 总上限
|
||
max_turns: int = 12 # 最多保留的对话轮次
|
||
max_memories: int = 5 # 单轮最多注入的记忆条数
|
||
memory_timeout_s: float = 0.3 # 检索超时，方案要求"记忆失败不阻塞对话"
|
||
|
||
|
||
# ---------------------------------------------------------------- 联网搜索
|
||
|
||
@dataclass
|
||
class SearchConfig:
|
||
"""SearXNG 搜索后端。
|
||
|
||
默认关闭，避免开发机没有 SearXNG 时改变普通对话行为；部署机显式设置
|
||
SEARCH_ENABLED=true。服务只需暴露在助手所在机器的回环地址。
|
||
"""
|
||
|
||
enabled: bool = field(default_factory=lambda: _env_bool("SEARCH_ENABLED", False))
|
||
base_url: str = field(
|
||
default_factory=lambda: _env("SEARXNG_BASE_URL", "http://127.0.0.1:18080"))
|
||
timeout_s: float = field(default_factory=lambda: _env_float("SEARCH_TIMEOUT_S", 6.0))
|
||
max_results: int = field(default_factory=lambda: _env_int("SEARCH_MAX_RESULTS", 5))
|
||
language: str = field(default_factory=lambda: _env("SEARCH_LANGUAGE", "zh-CN"))
|
||
|
||
|
||
# ---------------------------------------------------------------- 语音
|
||
|
||
def _default_model_dir() -> Path:
|
||
"""模型目录。可用环境变量 AI_MODEL_DIR 覆盖。
|
||
|
||
为什么要有这个覆盖：开发在 N5095，运行在服务器，两边模型放的
|
||
位置不一样（服务器上模型在 ~/ai_companion_bench/models，
|
||
不在代码目录里）。没有覆盖的话，部署就得改代码——
|
||
而"改代码才能部署"是部署流程里最容易出错的一环。
|
||
"""
|
||
override = _env("AI_MODEL_DIR")
|
||
if override:
|
||
return Path(override).expanduser()
|
||
return PROJECT_ROOT / "p0" / "stt_tts_bench" / "models"
|
||
|
||
|
||
@dataclass
|
||
class VoiceConfig:
|
||
"""语音输入输出。
|
||
|
||
线程数是实测出来的，不能按"核多给多"猜：
|
||
- N5095（4核，无AVX2）：STT 超过 2 线程急剧退化，4线程慢 3.9 倍
|
||
- 服务器（i9-14900K，有AVX2）：同样配置下 RTF 0.03
|
||
所以线程数按部署机器配置，不写死。服务器上可以给足。
|
||
"""
|
||
model_dir: Path = field(default_factory=_default_model_dir)
|
||
sample_rate: int = 16000
|
||
stt_threads: int = field(default_factory=lambda: _env_int("AI_STT_THREADS", 2))
|
||
tts_threads: int = field(default_factory=lambda: _env_int("AI_TTS_THREADS", 4))
|
||
|
||
# ---- TTS 后端选择 ----
|
||
# "local" = sherpa-onnx vits-zh-ll：RTF 0.09，很快，但音色用户不满意
|
||
# "remote" = Qwen3-TTS 独立服务：RTF 0.55–0.94，音质是另一档，用户试听选定
|
||
tts_backend: str = field(default_factory=lambda: _env("AI_TTS_BACKEND", "remote"))
|
||
tts_service_url: str = field(
|
||
default_factory=lambda: _env("TTS_SERVICE_URL", "http://127.0.0.1:18040"))
|
||
# 用户试听选定的组合（out/tts_qwen/instruct_7849.wav）
|
||
tts_speaker: str = field(default_factory=lambda: _env("AI_TTS_SPEAKER", "Serena"))
|
||
tts_instructions: str = field(
|
||
default_factory=lambda: _env("AI_TTS_INSTRUCTIONS", "用轻快活泼的语气说"))
|
||
tts_language: str = field(default_factory=lambda: _env("AI_TTS_LANGUAGE", "Chinese"))
|
||
|
||
|
||
# ---------------------------------------------------------------- 记忆
|
||
|
||
@dataclass
|
||
class MemoryConfig:
|
||
db_path: Path = field(default_factory=lambda: Path.home() / ".local" / "share" / "ai_assistant" / "memory.db")
|
||
# 自动抽取的记忆默认是"候选"，不参与检索；只有用户显式确认的才生效。
|
||
# 对应方案 §8："模型推测不当作确认事实"
|
||
auto_extract_enabled: bool = True
|
||
auto_extract_confirm_status: str = "candidate"
|
||
|
||
|
||
# ---------------------------------------------------------------- 汇总
|
||
|
||
@dataclass
|
||
class AgentConfig:
|
||
"""Agent 运行时的选择。
|
||
|
||
Phase 3 起有两条路径，**默认必须是 legacy**——交接文档要求
|
||
"flows 默认值只有在线上验收后才能从 legacy 改为 flows"。
|
||
"""
|
||
|
||
#: "legacy" = ConversationEngine 手写工具循环（当前线上行为）
|
||
#: "flows" = Pipecat Flows 驱动文字路径（Phase 3 新路径）
|
||
runtime: str = field(default_factory=lambda: _env("AI_AGENT_RUNTIME", "legacy"))
|
||
|
||
#: Flows 路径的 max_tokens。
|
||
#: **不能沿用语音的 200**：实测同一句"忘掉我的饮食偏好"，
|
||
#: max_tokens=200 时模型直接 finish=stop、根本不调工具；
|
||
#: 600 时才正常产生 tool_calls 且参数 JSON 完整。
|
||
#: 200 是为"语音回复要短"设的，而带工具的路径需要额外余量。
|
||
max_tokens: int = field(
|
||
default_factory=lambda: _env_int("AI_AGENT_MAX_TOKENS", 600))
|
||
|
||
@property
|
||
def use_flows(self) -> bool:
|
||
return self.runtime.strip().lower() == "flows"
|
||
|
||
|
||
@dataclass
|
||
class AppConfig:
|
||
# 单用户原型的稳定内部档案 ID。它只用于数据隔离，绝不是用户可见姓名。
|
||
default_user_id: str = field(
|
||
default_factory=lambda: _env("AI_DEFAULT_USER_ID", "u_ylhan"))
|
||
#: 主 LLM 由 AI_LLM_PRIMARY 选择：
|
||
#: "local" = 本机 vLLM（首 token 192ms，但占 21.5GB 显存）
|
||
#: "deepseek" = 云端（首 token 786ms，网络往返约 700ms）
|
||
#: 2026-09-15 起用 deepseek：本地 27B 已停，GPU 让给 Qwen3-TTS。
|
||
#: 这是个可逆的权衡——音质提升换首音延迟增加约 600ms。
|
||
llm: LLMConfig = field(default_factory=local_llm_config)
|
||
llm_fallback: LLMConfig = field(default_factory=deepseek_config)
|
||
context: ContextConfig = field(default_factory=ContextConfig)
|
||
search: SearchConfig = field(default_factory=SearchConfig)
|
||
voice: VoiceConfig = field(default_factory=VoiceConfig)
|
||
memory: MemoryConfig = field(default_factory=MemoryConfig)
|
||
agent: AgentConfig = field(default_factory=AgentConfig)
|
||
|
||
@classmethod
|
||
def load(cls) -> "AppConfig":
|
||
load_env()
|
||
primary = _env("AI_LLM_PRIMARY", "local").strip().lower()
|
||
if primary == "deepseek":
|
||
return cls(llm=deepseek_config(), llm_fallback=local_llm_config())
|
||
return cls()
|
||
|
||
|
||
__all__ = [
|
||
"AppConfig", "LLMConfig", "ContextConfig", "SearchConfig", "VoiceConfig", "MemoryConfig",
|
||
"load_env", "local_llm_config", "deepseek_config", "PROJECT_ROOT", "ENV_FILE",
|
||
]
|