diff --git a/app/agent/flows_runtime.py b/app/agent/flows_runtime.py new file mode 100644 index 0000000..1da78b8 --- /dev/null +++ b/app/agent/flows_runtime.py @@ -0,0 +1,350 @@ +"""Pipecat Flows 运行时装配（Phase 3）。 + +**只接管网页文字路径。** 语音仍走 legacy 的 `EngineProcessor`——交接文档 +明确要求 "语音路径暂时保持 legacy"，等 Phase 3、4 稳定后再迁。 + +管线形态： + + transport.input() + → EngineProcessor(owns_text=False) 语音仍由它处理；文字帧往下传 + → flows 的 context aggregator.user() + → DeepSeekLLMService Flows 驱动 + → FlowsBridge LLM 文本 → RTVI 消息 + 分句 TTS + → flows 的 context aggregator.assistant() + → transport.output() + +两个关键设计： + + * **只用 `task_messages`，不用 `role_message`。** Phase 1 spike 实测： + 两者会各产生一条 system 消息进请求。只用一种。这里把人格与任务都写进 + `task_messages`，与 legacy"每轮重建单条 system"的做法一致。 + * **工具执行复用 `app/agent/tools.py`**，不另写一套。身份由服务端注入， + 模型不得传。 +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime +from typing import Any, AsyncIterator, Callable + +import numpy as np + +from pipecat.frames.frames import ( + Frame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + TextFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +from ..config import AppConfig +from ..memory.base import MemoryStore +from ..prompts import PERSONA_RULES +from ..voice.tts import split_sentences +from . import tools as agent_tools + +log = logging.getLogger(__name__) + + +# ---------------------------------------------------------------- 本轮用户原话 + +#: 本轮用户原话。策略层要它区分"用户明确要求记住"和"模型自己推断"， +#: 而 Flows 的工具 handler 只看得到工具参数，看不到原话。 +#: 用单元素列表做可变闭包——handler 是闭包，拿不到外层的非局部变量。 +#: +#: 注意：这是**单用户在用的原型**。多用户并发会话时这里会互相覆盖， +#: 要支持并发得改成按 session 分键（Phase 5 迁移语音路径时一并处理）。 +_LAST_USER_TEXT: list[str] = [""] + + +def set_last_user_text(text: str) -> None: + _LAST_USER_TEXT[0] = text or "" + + +# ---------------------------------------------------------------- 节点 + +#: 会话控制状态。**不要把长期记忆复制进来**——Flows 的 state 不是数据库， +#: 长期数据一律走 MemoryStore（交接文档 §七）。 +def _conversation_node(*, display_name: str, tool_schemas: list[dict], + ctx_factory: Callable[[], agent_tools.ToolContext]) -> dict: + """默认节点：普通陪伴对话 + 可用工具。 + + 人格与任务都放在 task_messages 里——**不用 role_message**， + 避免两条 system 消息（见模块文档）。 + """ + persona = PERSONA_RULES + if display_name: + persona += f"\n\n当前用户希望被称为：{display_name}。" + + from pipecat.flows import FlowsFunctionSchema + + functions = [] + for schema in tool_schemas: + fn = schema["function"] + functions.append(FlowsFunctionSchema( + name=fn["name"], + description=fn["description"], + properties=fn["parameters"].get("properties", {}), + required=fn["parameters"].get("required", []), + handler=_make_handler(fn["name"], ctx_factory), + )) + + return { + "name": "conversation", + "task_messages": [ + {"role": "system", "content": persona}, + ], + "functions": functions, + # 进入节点不自动开口：文字路径要等用户先说话。 + # （spike 实测默认 True 会在 initialize 时立刻触发一轮回复， + # 那条回复与用户输入无关，会污染归因，也会让网页莫名多一段话。） + "respond_immediately": False, + } + + +def _make_handler(name: str, ctx_factory: Callable[[], agent_tools.ToolContext]): + """把工具层的执行器包成 Flows 的 handler。 + + **用闭包捕获 ctx_factory，不用模块级全局**——全局在并发连接时会串 + （两个会话各自的 owner_user_id 会互相覆盖）。 + """ + async def handler(args, flow_manager=None): + outcome = await agent_tools.execute(name, dict(args or {}), ctx_factory()) + return outcome.content + return handler + + +# ---------------------------------------------------------------- 输入记录 + +class UserTextRecorder(FrameProcessor): + """记录本轮用户原话，供策略层判定语义等级。 + + 策略层要区分"用户明确要求记住"和"模型自己推断"，依据是**用户原话**， + 而 Flows 的工具 handler 只看得到工具参数。所以这里把原话截下来。 + """ + + def __init__(self, extractor): + super().__init__() + self._extract = extractor + + async def process_frame(self, frame: Frame, direction: FrameDirection): + from pipecat.frames.frames import LLMMessagesAppendFrame + await super().process_frame(frame, direction) + if isinstance(frame, LLMMessagesAppendFrame): + try: + text = self._extract(frame.messages) + if text: + set_last_user_text(text) + except Exception: # noqa: BLE001 — 记录失败不该影响对话 + log.debug("记录用户原话失败", exc_info=True) + await self.push_frame(frame, direction) + + +# ---------------------------------------------------------------- 输出桥 + +class FlowsBridge(FrameProcessor): + """把 Flows/LLM 的文本帧转成 RTVI 消息 + 分句 TTS。 + + 产出**与 legacy 的 `_run_turn` 保持一致**（网页依赖这些消息渲染）： + bot-llm-started → 逐 token 的 bot-llm-text → bot-llm-stopped + → bot-transcription / bot-output（完整文本） + """ + + def __init__(self, *, rtvi, tts, cfg: AppConfig, audio_enabled: bool = True): + super().__init__() + self.rtvi = rtvi + self.tts = tts + self.cfg = cfg + self.audio_enabled = audio_enabled + self._buffer = "" + self._reply: list[str] = [] + self._started = False + self._speak_tasks: list[asyncio.Task] = [] + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, LLMFullResponseStartFrame): + self._started = True + self._buffer = "" + self._reply = [] + await self.push_frame(frame, direction) + await self.rtvi.push_transport_message(_rtvi_bot_started()) + + elif isinstance(frame, TextFrame) and frame.text: + self._buffer += frame.text + self._reply.append(frame.text) + # 逐 token 给网页（Playground 的 Conversation 视图需要） + await self.rtvi.push_transport_message(_rtvi_bot_text(frame.text)) + await self.push_frame(frame, direction) + # 攒到句子边界就送去合成——这是提前出声的唯一手段 + segs = split_sentences(self._buffer) + if len(segs) > 1: + *ready, self._buffer = segs + for seg in ready: + await self._speak(seg) + + elif isinstance(frame, LLMFullResponseEndFrame): + if self._buffer.strip(): + await self._speak(self._buffer) + self._buffer = "" + reply = "".join(self._reply).strip() + if reply: + # 完整文本单发一次：Playground 1.7 的 Conversation 视图 + # 主要消费完整转写/bot-output，只给 token 级不会稳定渲染。 + await self.rtvi.push_transport_message(_rtvi_bot_transcription(reply)) + await self.rtvi.push_transport_message( + _rtvi_bot_output(reply, spoken=self.audio_enabled)) + if self._started: + await self.rtvi.push_transport_message(_rtvi_bot_stopped()) + self._started = False + await self.push_frame(frame, direction) + + else: + await self.push_frame(frame, direction) + + async def _speak(self, text: str) -> None: + text = (text or "").strip() + if not text or not self.audio_enabled: + return + try: + audio = await asyncio.to_thread(self.tts.synthesize, text) + except Exception as e: # noqa: BLE001 — TTS 失败不该中断文字显示 + log.warning("Flows 路径 TTS 失败（文字仍显示）：%s", e) + return + if audio is None or not len(audio): + return + await self._push_audio(np.asarray(audio, dtype=np.float32), + getattr(self.tts, "sample_rate", 24000)) + + async def _push_audio(self, samples: np.ndarray, sample_rate: int) -> None: + from pipecat.frames.frames import OutputAudioRawFrame + out_rate = self.cfg.voice.output_sample_rate if hasattr( + self.cfg.voice, "output_sample_rate") else 24000 + if sample_rate != out_rate: + n = int(len(samples) * out_rate / sample_rate) + samples = np.interp(np.linspace(0, len(samples) - 1, n), + np.arange(len(samples)), samples).astype(np.float32) + pcm = (np.clip(samples, -1, 1) * 32767).astype(np.int16) + step = out_rate // 50 # 20ms 一块 + for i in range(0, len(pcm), step): + await self.push_frame(OutputAudioRawFrame( + audio=pcm[i:i + step].tobytes(), sample_rate=out_rate, + num_channels=1)) + + +# ---------------------------------------------------------------- RTVI 消息 + +def _rtvi_bot_started(): + import pipecat.processors.frameworks.rtvi.models as R + return R.BotLLMStartedMessage() + + +def _rtvi_bot_stopped(): + import pipecat.processors.frameworks.rtvi.models as R + return R.BotLLMStoppedMessage() + + +def _rtvi_bot_text(text: str): + import pipecat.processors.frameworks.rtvi.models as R + return R.BotLLMTextMessage(data=R.TextMessageData(text=text)) + + +def _rtvi_bot_transcription(text: str): + import pipecat.processors.frameworks.rtvi.models as R + return R.BotTranscriptionMessage(data=R.TextMessageData(text=text)) + + +def _rtvi_bot_output(text: str, *, spoken: bool): + import pipecat.processors.frameworks.rtvi.models as R + return R.BotOutputMessage(data=R.BotOutputMessageData( + text=text, aggregated_by="sentence", spoken=spoken)) + + +# ---------------------------------------------------------------- 装配 + +async def build_flows_parts( + *, + cfg: AppConfig, + memory: MemoryStore, + identity, + session_id: str, + tts, + rtvi, + audio_enabled: bool = True, +): + """造出 Flows 路径需要的**零件**，不含 FlowManager。 + + 为什么不在这里建 FlowManager：它要求传 `worker`（PipelineWorker）， + 而 worker 又必须在管线建好之后才能建，是个先后依赖。 + 所以顺序是：零件 → 管线 → worker → FlowManager → initialize。 + + 返回 (llm, aggregator, bridge, ctx_factory, tool_schemas) + """ + from .llm_service import create_llm_service + + llm = create_llm_service(cfg.llm) + if llm is None: + raise RuntimeError( + f"Flows 路径不支持后端 {cfg.llm.name!r}（只支持 deepseek）。" + f"请把 AI_AGENT_RUNTIME 改回 legacy。") + + context = LLMContext() + aggregator = LLMContextAggregatorPair(context) + + def ctx_factory() -> agent_tools.ToolContext: + """每次工具调用现算身份——用户可能在会话中途切换。""" + d = identity.current() + return agent_tools.ToolContext( + owner_user_id=d.user_id, # ← 服务端注入，模型不得传 + session_id=session_id, + turn_id=f"t_{datetime.now(UTC).strftime('%H%M%S%f')}", + identity_confirmed=d.confirmed, + user_text=_LAST_USER_TEXT[0], + store=memory, + # Phase 3 先不暴露 web_search：搜索仍走 legacy 路径， + # 等 flows 文字路径验收通过后再接（交接文档要求分阶段）。 + search=None, + memory_limit=cfg.context.max_memories, + memory_timeout_s=cfg.context.memory_timeout_s, + ) + + bridge = FlowsBridge(rtvi=rtvi, tts=tts, cfg=cfg, audio_enabled=audio_enabled) + tool_schemas = agent_tools.available_schemas(ctx_factory()) + return llm, aggregator, bridge, ctx_factory, tool_schemas + + +def make_flow_manager(*, worker, llm, aggregator): + """worker 建好之后再建 FlowManager。""" + from pipecat.flows import FlowManager + return FlowManager(worker=worker, llm=llm, context_aggregator=aggregator) + + +async def initialize_conversation_node( + flow, *, display_name: str, tool_schemas: list[dict], + ctx_factory: Callable[[], agent_tools.ToolContext], +) -> None: + """设置初始节点。 + + **必须 await。** `FlowManager.initialize` 是协程，但签名显示 `-> None`， + 漏 await 不报错，只是人格/任务/工具**全部静默失效**—— + Phase 1 spike 在这里栽过一次，表现极具误导性： + 模型会正常回复（因为 LLM 照样被调用），但没有人格、没有工具。 + """ + node = _conversation_node(display_name=display_name, + tool_schemas=tool_schemas, + ctx_factory=ctx_factory) + await flow.initialize(node) + + +__all__ = [ + "FlowsBridge", "UserTextRecorder", "build_flows_parts", "make_flow_manager", + "initialize_conversation_node", + "set_last_user_text", "PERSONA_RULES", +] diff --git a/app/config.py b/app/config.py index 6336c71..4fa5aff 100644 --- a/app/config.py +++ b/app/config.py @@ -212,6 +212,23 @@ class MemoryConfig: # ---------------------------------------------------------------- 汇总 +@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")) + + @property + def use_flows(self) -> bool: + return self.runtime.strip().lower() == "flows" + + @dataclass class AppConfig: # 单用户原型的稳定内部档案 ID。它只用于数据隔离，绝不是用户可见姓名。 @@ -228,6 +245,7 @@ class AppConfig: 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": diff --git a/app/tests/test_flows_runtime.py b/app/tests/test_flows_runtime.py new file mode 100644 index 0000000..d171bf0 --- /dev/null +++ b/app/tests/test_flows_runtime.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python3 +"""Phase 3 Flows 运行时装配测试。 + +**不依赖网络、不启动 WebRTC**。验证的是装配正确性： + + - 节点配置形态（只用 task_messages、不自动开口） + - 工具 schema 随身份变化 + - handler 桥接到工具层，且身份由服务端注入 + - 运行时开关默认是 legacy + +真正跑起来的行为（RTVI 消息、TTS、延迟）由现场测试和真人网页验收覆盖。 + +用法： + ~/venvs/ai_assistant/bin/python app/tests/test_flows_runtime.py +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(ROOT)) + +from app.agent import flows_runtime as FR # noqa: E402 +from app.agent.tools import ToolContext # noqa: E402 +from app.config import AgentConfig, AppConfig # noqa: E402 +from app.identity import ManualIdentityProvider # noqa: E402 +from app.memory.base import ConfirmStatus, Memory, MemoryKind # noqa: E402 + +OK, FAIL = [], [] + + +def check(name: str, cond: bool, detail: str = "") -> None: + (OK if cond else FAIL).append(name) + print(f" [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) + + +class FakeStore: + """最小实现，只够装配与 handler 调用。""" + def __init__(self): + self.rows: list[Memory] = [] + self._n = 1 + + async def retrieve(self, owner_user_id, query, **kw): + return [r for r in self.rows if r.owner_user_id == owner_user_id][:5] + + async def remember(self, mem: Memory) -> int: + m = Memory(**{**mem.__dict__, "memory_id": self._n}) + self._n += 1 + self.rows.append(m) + return m.memory_id + + async def forget(self, owner_user_id, memory_id): + before = len(self.rows) + self.rows = [r for r in self.rows + if not (r.memory_id == memory_id + and r.owner_user_id == owner_user_id)] + return len(self.rows) < before + + +class FakeTTS: + sample_rate = 24000 + def synthesize(self, text): + import numpy as np + return np.zeros(2400, dtype=np.float32) + + +class FakeRTVI: + def __init__(self): self.msgs = [] + async def push_transport_message(self, m): self.msgs.append(type(m).__name__) + + +# ---------------------------------------------------------------- 配置开关 + +def test_runtime_switch(): + print("\n=== 1. 运行时开关 ===") + check("默认是 legacy", AgentConfig(runtime="legacy").use_flows is False) + check("显式 flows 才启用", AgentConfig(runtime="flows").use_flows is True) + check("大小写不敏感", AgentConfig(runtime="FLOWS").use_flows is True) + check("未知值当 legacy", AgentConfig(runtime="xyz").use_flows is False) + + saved = os.environ.pop("AI_AGENT_RUNTIME", None) + try: + from app.config import load_env + import importlib + import app.config as C + C._loaded = False # 重新读一次环境 + check("未设环境变量时默认 legacy", + AppConfig.load().agent.runtime == "legacy") + finally: + if saved is not None: + os.environ["AI_AGENT_RUNTIME"] = saved + + +# ---------------------------------------------------------------- 节点形态 + +def test_node_shape(): + print("\n=== 2. 节点配置形态 ===") + identity = ManualIdentityProvider("u_a") + store = FakeStore() + bridge = FR.FlowsBridge(rtvi=FakeRTVI(), tts=FakeTTS(), + cfg=AppConfig.load()) + + def ctx_factory(): + d = identity.current() + return ToolContext(owner_user_id=d.user_id, session_id="s1", + turn_id="t1", identity_confirmed=d.confirmed, + store=store) + + schemas = [ + {"type": "function", "function": { + "name": "memory_search", "description": "x", + "parameters": {"type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"]}}}, + ] + node = FR._conversation_node(display_name="林", tool_schemas=schemas, + ctx_factory=ctx_factory) + + check("节点名是 conversation", node["name"] == "conversation") + check("**不用 role_message**", "role_message" not in node, + "避免两条 system 消息") + check("用 task_messages 承载人格", + len(node["task_messages"]) == 1 + and node["task_messages"][0]["role"] == "system") + check("人格里含称呼", "林" in node["task_messages"][0]["content"]) + check("respond_immediately=False（不自动开口）", + node["respond_immediately"] is False) + check("工具已注册", len(node["functions"]) == 1) + + +# ---------------------------------------------------------------- handler 桥接 + +async def test_handler_bridge(): + print("\n=== 3. handler 桥接到工具层 ===") + identity = ManualIdentityProvider("u_a") + store = FakeStore() + FR.set_last_user_text("记住我不吃香菜") + + def ctx_factory(): + d = identity.current() + return ToolContext(owner_user_id=d.user_id, session_id="s1", + turn_id="t1", identity_confirmed=d.confirmed, + user_text=FR._LAST_USER_TEXT[0], store=store) + + handler = FR._make_handler("memory_remember", ctx_factory) + out = await handler({"kind": "preference", "key": "饮食偏好", + "value": "不吃香菜"}) + check("handler 返回文本", isinstance(out, str) and "已记住" in out, out[:40]) + check("写入的 owner 来自服务端", + store.rows and store.rows[0].owner_user_id == "u_a") + + # 身份未确认时，handler 应拒绝（策略层兜底） + identity.reset() + out2 = await handler({"kind": "preference", "value": "x"}) + # 策略层给模型的话是"现在还不能记录私人信息。请先确认身份。"， + # 断言要匹配实际措辞（早先写死"身份未确认"导致误判成 FAIL）。 + check("身份未确认时 handler 拒绝", + "不能记录" in out2 or "确认身份" in out2, out2[:40]) + + # 越权参数应被工具层拦下 + identity.switch("u_a") + out3 = await handler({"kind": "fact", "value": "x", "owner_user_id": "u_b"}) + check("越权参数被拦", "不合法" in out3, out3[:40]) + + +# ---------------------------------------------------------------- 零件装配 + +async def test_build_parts(): + print("\n=== 4. build_flows_parts 装配 ===") + cfg = AppConfig.load() + # 强制用 deepseek 配置（本地 env 可能没设 AI_LLM_PRIMARY） + from app.config import deepseek_config + cfg.llm = deepseek_config() + + if not cfg.llm.api_key: + check("跳过（无 DeepSeek 密钥）", True) + return + + llm, aggregator, bridge, ctx_factory, schemas = await FR.build_flows_parts( + cfg=cfg, memory=FakeStore(), identity=ManualIdentityProvider("u_a"), + session_id="s1", tts=FakeTTS(), rtvi=FakeRTVI()) + + check("返回 LLM service", llm is not None and type(llm).__name__ == "DeepSeekLLMService") + check("模型名被覆盖为 deepseek-flash", + llm._settings.model == cfg.llm.model, llm._settings.model) + check("thinking 关闭", + getattr(llm._settings.thinking, "type", None) == "disabled") + check("aggregator 有 user/assistant", + aggregator.user() is not None and aggregator.assistant() is not None) + check("工具 schema 随身份给出（含记忆工具）", + any(s["function"]["name"].startswith("memory_") for s in schemas), + str([s["function"]["name"] for s in schemas])) + check("Phase 3 不含 web_search", + not any(s["function"]["name"] == "web_search" for s in schemas), + "搜索仍走 legacy 路径") + + # 身份未确认时不暴露记忆工具 + _, _, _, _, schemas2 = await FR.build_flows_parts( + cfg=cfg, memory=FakeStore(), identity=ManualIdentityProvider(), + session_id="s1", tts=FakeTTS(), rtvi=FakeRTVI()) + check("身份未确认时不给记忆工具", + not any(s["function"]["name"].startswith("memory_") for s in schemas2)) + + +def main() -> int: + test_runtime_switch() + test_node_shape() + asyncio.run(test_handler_bridge()) + asyncio.run(test_build_parts()) + + print("\n" + "=" * 62) + print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}") + for n in FAIL: + print(f" 未通过: {n}") + return 0 if not FAIL else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/client/test_client.py b/client/test_client.py index 4ff7852..b734034 100644 --- a/client/test_client.py +++ b/client/test_client.py @@ -191,6 +191,26 @@ class ReplyRecorder: return (self.first_non_silent_at - started_at) * 1000 +def rtvi_send_text(channel, text: str, *, audio_response: bool = True) -> None: + """通过 data channel 发一条 RTVI send-text。 + + 信封格式来自 pipecat 的 RTVI 模型（实测校正过两次）： + label='rtvi-ai'，**字段是 id 不是 msg_id**，type='send-text'， + data.content（**是 content 不是 text**）。 + 写错任何一个，服务端都会以 "Invalid RTVI transport message: Field required" 拒掉， + 而客户端看不到任何反馈——表现为"发了消息但助手毫无反应"。 + """ + import uuid + msg = { + "label": "rtvi-ai", + "id": uuid.uuid4().hex, + "type": "send-text", + "data": {"content": text, + "options": {"audio_response": audio_response}}, + } + channel.send(json.dumps(msg)) + + def post_offer(server: str, sdp: str, type_: str) -> dict: """把 offer 发给服务端，取回 answer。""" payload = json.dumps({"sdp": sdp, "type": type_, "pc_id": None, @@ -203,7 +223,7 @@ def post_offer(server: str, sdp: str, type_: str) -> dict: async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, - timeout_s: float) -> int: + timeout_s: float, texts: list[str] | None = None) -> int: pc = RTCPeerConnection(RTCConfiguration(iceServers=[])) recorder = ReplyRecorder(out) track = None @@ -235,6 +255,10 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, track = FileAudioTrack(wav) pc.addTrack(track) log.info("音源: %s（%d 采样点）", wav.name, len(track.samples)) + elif texts: + # 文字模式：仍要一条音轨，否则 aiortc 不收集 ICE 候选 + pc.addTrack(SilentTrack()) + log.info("文字模式：将发送 %d 条文本", len(texts)) else: # --connect-only 也要加一条（静音）音轨。 # **不能省**：aiortc 的 offer 里没有媒体时完全不收集 ICE 候选， @@ -312,7 +336,31 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, log.info("✅ ICE 已连通（%s）", pc.connectionState) print(f" ✅ 连接建立，状态 {pc.connectionState}") - if track: + if texts: + # 等 data channel 就绪（服务端靠它判断客户端存活） + for _ in range(60): + if dc_state["ok"]: + break + await asyncio.sleep(0.1) + log.info("data channel: %s", "已就绪" if dc_state["ok"] else "未就绪") + for t in texts: + await asyncio.sleep(0.5) + rtvi_send_text(channel, t) + log.info("已发送文本: %r", t) + # 等**非静音**音频出现。 + # 不能用"收到多少块"来判断——WebRTC 输出轨空闲时也在推静音， + # 块数会立刻涨上去，导致客户端在服务端还在生成时就断开 + # （服务端日志表现为"本轮被取消"，客户端表现为"没反应"）。 + recorder.first_non_silent_at = None + for _ in range(int(timeout_s * 10)): + if recorder.has_speech: + break + await asyncio.sleep(0.1) + log.info("本轮 %s", "收到语音回复" if recorder.has_speech else "未收到语音") + # 让它把话说完 + await asyncio.sleep(4) + await asyncio.sleep(2) + elif track: # 等音频放完 + 留出响应时间 t0 = time.perf_counter() while not track.finished and time.perf_counter() - t0 < timeout_s: @@ -350,13 +398,16 @@ def main() -> int: p.add_argument("--wav", type=Path) p.add_argument("--out", type=Path, default=Path("/tmp/client_reply.wav")) p.add_argument("--connect-only", action="store_true") + p.add_argument("--text", action="append", default=None, + help="通过 RTVI 发送文本（可重复），用于测 flows 文字路径") p.add_argument("--timeout", type=float, default=30.0) a = p.parse_args() - if not a.connect_only and not a.wav: - print("需要 --wav 或 --connect-only", file=sys.stderr) + if not a.connect_only and not a.wav and not a.text: + print("需要 --wav / --text / --connect-only", file=sys.stderr) return 2 return asyncio.run(run(a.server, a.wav, a.out, - connect_only=a.connect_only, timeout_s=a.timeout)) + connect_only=a.connect_only, timeout_s=a.timeout, + texts=a.text)) if __name__ == "__main__": diff --git a/deploy.sh b/deploy.sh new file mode 100755 index 0000000..867ca5b --- /dev/null +++ b/deploy.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# 部署到服务器。 +# +# **为什么要有这个脚本**：2026-09-16 我用一条 rsync 命令把服务器上的东西 +# 删了一大片，教训有三条，都固化在这里： +# +# 1. **绝不用 `--delete`。** 服务器上还有本地没有的东西（.venv、models 软链、 +# logs、数据库软链等）。`--delete` 会把它们全删掉。 +# 实测后果：.venv/bin 与 100+ 个 *.dist-info 被删、services/ 被清空、 +# README/部署说明被删、models 软链被删。 +# 2. **一次只同步一个目录，不要多源。** 多源 rsync 不带 --relative 时， +# 会把 `app/` 的内容**平铺**到目标根目录，而不是放进 `app/` 下。 +# 3. **排除清单要显式。** 即使不用 --delete，也别把 __pycache__ 之类带过去。 +# +# 用法： +# ./deploy.sh # 同步代码 +# ./deploy.sh --restart # 同步后重启两个服务 +set -uo pipefail + +HOST="${AI_DEPLOY_HOST:-yonglin@cadadmin-server.taild0f91d.ts.net}" +DEST="${AI_DEPLOY_DEST:-~/ai-companion}" +HERE="$(cd "$(dirname "$0")" && pwd)" + +# 服务器上不该被本地覆盖的东西——保留在服务器上 +EXCLUDES=( + --exclude '__pycache__' + --exclude '*.pyc' + --exclude '.baseline' + --exclude '.venv' + --exclude 'logs' + --exclude 'models' + --exclude 'out' + --exclude 'p0/stt_tts_bench/models' +) + +sync_one() { # sync_one <本地目录> <远端子路径> + echo " → $2" + rsync -az "${EXCLUDES[@]}" -e "ssh -o BatchMode=yes" \ + "$HERE/$1" "$HOST:$DEST/$2" +} + +echo "部署到 $HOST:$DEST" +sync_one "app/" "app/" +sync_one "services/" "services/" +sync_one "client/" "client/" +sync_one "p1/" "p1/" + +for f in README.md 部署说明.md requirements.txt; do + [ -f "$HERE/$f" ] && rsync -az -e "ssh -o BatchMode=yes" "$HERE/$f" "$HOST:$DEST/" +done + +# 把 requirements 里没有、但运行需要的包装上（幂等） +ssh -o BatchMode=yes "$HOST" "cd $DEST && .venv/bin/pip -q install pipecat-ai-small-webrtc-prebuilt 2>&1 | tail -1" + +if [ "${1:-}" = "--restart" ]; then + echo "重启服务" + ssh -o BatchMode=yes "$HOST" 'bash -s' </dev/null); do kill "\$p" 2>/dev/null; done +sleep 3 +cd ~/tts-qwen && setsid nohup .venv/bin/python tts_server.py --port 18040 > logs/tts_service.log 2>&1 < /dev/null & +sleep 1 +cd ~/ai-companion && setsid nohup .venv/bin/python services/bot_server.py --host 0.0.0.0 --port 7860 > logs/bot_server.log 2>&1 < /dev/null & +for i in \$(seq 1 30); do + curl -sf -o /dev/null http://127.0.0.1:7860/health 2>/dev/null && { echo " bot ✅"; break; } + sleep 3 +done +curl -sf http://127.0.0.1:18040/health 2>/dev/null | grep -q model_loaded && echo " tts ✅" || echo " tts ⚠️" +REMOTE +fi + +echo "完成" diff --git a/services/bot_server.py b/services/bot_server.py index 896d264..5da04f1 100644 --- a/services/bot_server.py +++ b/services/bot_server.py @@ -59,6 +59,7 @@ from pipecat.frames.frames import ( TranscriptionFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -75,6 +76,10 @@ from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport from pipecat.utils.asyncio.task_manager import TaskManager from pipecat.workers.base_worker import WorkerParams +from app.agent.flows_runtime import ( + UserTextRecorder, build_flows_parts, initialize_conversation_node, + make_flow_manager, +) from app.config import AppConfig from app.engine import ConversationEngine, EventKind from app.identity import ManualIdentityProvider @@ -199,11 +204,17 @@ class EngineProcessor(FrameProcessor): cfg: AppConfig, transport: SmallWebRTCTransport, rtvi: RTVIProcessor, + *, + owns_text: bool = True, ): super().__init__() self.cfg = cfg self.transport = transport self.rtvi = rtvi + #: True = 文字路径也归本处理器（legacy） + #: False = 文字帧往下传给 Flows 处理器（Phase 3 的 flows 运行时） + #: 语音路径**始终**归本处理器——交接文档要求语音暂不迁移。 + self.owns_text = owns_text self._vad = SimpleVAD() self._pre_roll: deque[bytes] = deque(maxlen=50) # 约 1 秒 self._buf = bytearray() # 当前话语的 PCM16 @@ -240,6 +251,15 @@ class EngineProcessor(FrameProcessor): self._engine.stt = StreamingSTT(self.cfg.voice) return self._engine + async def runtime_deps(self): + """暴露 memory/identity/session/tts，供 Flows 路径复用。 + + 资源只在一个地方创建（`_ensure_engine`），避免两条路径各建一份 + SQLite 连接或各加载一次 TTS——那会立刻导致"两条路径看到不同数据"。 + """ + engine = await self._ensure_engine() + return self._memory, engine.identity, self._session, engine.tts + async def setup(self, setup): await super().setup(setup) log.info("会话处理器已就绪") @@ -255,10 +275,17 @@ class EngineProcessor(FrameProcessor): elif isinstance(frame, LLMConfigureOutputFrame): # RTVI 在 send-text 前后用这个帧表达 audio_response 选项。 + # 即使在 flows 模式下也记录一份（语音路径仍归本处理器）， + # 同时往下传，让 FlowsBridge 也能看到。 self._skip_tts = frame.skip_tts await self.push_frame(frame, direction) elif isinstance(frame, LLMMessagesAppendFrame): + if not self.owns_text: + # flows 运行时：文字交给下游的 Flows 处理器， + # 这里只放行，不消费。 + await self.push_frame(frame, direction) + return text = extract_user_text(frame.messages) if text and frame.run_llm is not False: log.info("收到网页文字输入: %d 字", len(text)) @@ -506,8 +533,35 @@ async def offer(request: SmallWebRTCRequest): ), ) rtvi = RTVIProcessor() - engine_proc = EngineProcessor(_cfg, transport, rtvi) - pipeline = Pipeline([transport.input(), engine_proc, transport.output()]) + # Phase 3：文字路径可选走 Flows，**语音始终归 EngineProcessor** + # （交接文档要求语音暂不迁移）。 + # 默认 legacy；只有显式设 AI_AGENT_RUNTIME=flows 才会走新路径。 + use_flows = _cfg.agent.use_flows + engine_proc = EngineProcessor(_cfg, transport, rtvi, owns_text=not use_flows) + + flows_objs = None + if use_flows: + # 复用 EngineProcessor 已建好的资源，不另建一份 + memory, identity, session_id, tts = await engine_proc.runtime_deps() + llm, aggregator, bridge, ctx_factory, tool_schemas = \ + await build_flows_parts( + cfg=_cfg, memory=memory, identity=identity, + session_id=session_id, tts=tts, rtvi=rtvi) + pipeline = Pipeline([ + transport.input(), + engine_proc, # 只吃语音 + UserTextRecorder(extract_user_text), # 记下用户原话供策略层 + aggregator.user(), # 收文字 + llm, # Flows 驱动 + bridge, # 文本 → RTVI + TTS + aggregator.assistant(), + transport.output(), + ]) + flows_objs = (llm, aggregator, ctx_factory, tool_schemas) + log.info("Agent 运行时: flows（文字路径），语音仍走 legacy") + else: + pipeline = Pipeline([transport.input(), engine_proc, transport.output()]) + log.info("Agent 运行时: legacy（默认）") # 注意：这个版本里 PipelineParams 没有 allow_interruptions—— # 打断在 1.10 已移到 pipecat/turns/ 框架。实验阶段我们不用它的 # 轮次框架（方案 §5 的边界决定是"打断由客户端判定"）， @@ -568,12 +622,33 @@ async def offer(request: SmallWebRTCRequest): async def _run_task(): log.info("PipelineWorker 开始运行") try: - # TaskManager 必须在这里建——模块级没有运行中的事件循环 - await task.run(WorkerParams(task_manager=TaskManager())) + # **必须用 PipelineRunner，不能直接 task.run(...)。** + # 直接 run 时 runner 没挂上，Pipecat 自己的函数调用机制会失败： + # "Worker 'PipelineWorker#N': runner is not set; call attach() first." + # 症状是工具被调用但 handler 从不执行，LLM 只收到一句通用的 + # "The function `X` failed and returned no result."，默认日志看不到。 + # legacy 路径不受影响（它的工具循环在自有引擎里）， + # 但 Flows 路径的工具走 Pipecat 机制，必须 attach。 + # 见笔记库 P1-Flows兼容性Spike结论.md 坑 2。 + await PipelineRunner(handle_sigint=False).run(task) except Exception as e: # noqa: BLE001 log.exception("PipelineWorker 异常退出: %s", e) log.info("PipelineWorker 已结束") + if flows_objs is not None: + llm, aggregator, ctx_factory, tool_schemas = flows_objs + flow = make_flow_manager(worker=task, llm=llm, aggregator=aggregator) + identity = engine_proc._engine.identity if engine_proc._engine else None + display_name = "" + if identity is not None: + display_name = getattr(identity.current(), "display_name", "") or "" + # **必须 await**：initialize 是协程但签名显示 -> None， + # 漏 await 不报错，只是人格/任务/工具全部静默失效（Phase 1 栽过）。 + await initialize_conversation_node( + flow, display_name=display_name, tool_schemas=tool_schemas, + ctx_factory=ctx_factory) + log.info("Flows 节点已初始化：工具 %d 个", len(tool_schemas)) + bg = asyncio.create_task(_run_task()) _bg_tasks.add(bg) bg.add_done_callback(_bg_tasks.discard)