"""Transportless production runner for one Responses API turn.""" from __future__ import annotations import asyncio from copy import deepcopy import json import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any from db.session import SessionLocal from loguru import logger from models import AssistantConfig from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.frames.frames import ( BotStoppedSpeakingFrame, CancelFrame, DataFrame, EndFrame, ErrorFrame, FunctionCallInProgressFrame, FunctionCallResultFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMTextFrame, ManuallySwitchServiceFrame, TTSSpeakFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.worker import PipelineParams, PipelineWorker 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 pipecat.services.llm_service import FunctionCallParams from pipecat.workers.runner import WorkerRunner from services.brains import BrainRuntime, build_brain from services.client_tools import ClientToolResponseWaitMode from services.knowledge import search as search_knowledge from services.openai_responses.protocol import new_id from services.pipecat.call_lifecycle import CallEndCoordinator from services.pipecat.processors import ( KnowledgeRetrievalProcessor, UserTurnRoutingProcessor, ) from services.pipecat.service_factory import config_with_resource from services.pipecat.workflow_services import build_workflow_llm_switcher IDLE_SETTLE_SECONDS = 0.2 WORKER_START_TIMEOUT_SECONDS = 5.0 class ResponseRunError(RuntimeError): pass DeltaCallback = Callable[[str], Awaitable[None]] @dataclass(frozen=True) class ResponseRunResult: text: str function_calls: list[dict[str, Any]] context_messages: list[dict[str, Any]] dynamic_variables: dict[str, Any] runtime_state: dict[str, Any] @dataclass class _ResponseWindow: active_llm: int = 0 active_assistant: int = 0 pending_tools: set[str] = field(default_factory=set) activity_count: int = 0 last_activity_at: float = field(default_factory=time.monotonic) def touch(self) -> None: self.activity_count += 1 self.last_activity_at = time.monotonic() @property def idle(self) -> bool: return self.active_llm == 0 and self.active_assistant == 0 and not self.pending_tools @dataclass class _BoundaryFrame(DataFrame): reached: asyncio.Event class _CaptureProcessor(FrameProcessor): def __init__(self, on_delta: DeltaCallback | None) -> None: super().__init__() self.on_delta = on_delta self.window = _ResponseWindow() self.deltas: list[str] = [] self.outputs: list[str] = [] self.error: str | None = None def reset(self) -> None: self.deltas.clear() self.outputs.clear() self.error = None self.window.activity_count = 0 self.window.last_activity_at = time.monotonic() async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, LLMFullResponseStartFrame): self.window.active_llm += 1 self.window.touch() elif isinstance(frame, LLMTextFrame): if frame.text: self.deltas.append(frame.text) self.window.touch() if self.on_delta is not None: await self.on_delta(frame.text) elif isinstance(frame, LLMFullResponseEndFrame): self.window.active_llm = max(0, self.window.active_llm - 1) self.window.touch() await self.push_frame(frame, direction) await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) return elif isinstance(frame, FunctionCallInProgressFrame): self.window.pending_tools.add(frame.tool_call_id) self.window.touch() elif isinstance(frame, FunctionCallResultFrame): self.window.pending_tools.discard(frame.tool_call_id) self.window.touch() elif isinstance(frame, TTSSpeakFrame): text = frame.text.strip() if text: self.outputs.append(text) self.window.touch() return elif isinstance(frame, ErrorFrame): self.error = frame.error or "Pipeline execution failed" self.window.pending_tools.clear() self.window.touch() elif isinstance(frame, _BoundaryFrame): frame.reached.set() self.window.touch() elif isinstance(frame, (EndFrame, CancelFrame)): self.window.touch() await self.push_frame(frame, direction) class ResponsesClientToolPort: """Turn a configured client-side tool invocation into a Response item.""" def __init__(self, results: list[dict[str, Any]] | None = None) -> None: self.calls: list[dict[str, Any]] = [] self.called = asyncio.Event() self._results = list(results or []) def supports(self, _function_name: str) -> bool: return True async def call( self, function_name: str, arguments: dict[str, Any], *, timeout_seconds: float, wait_for_response: bool = True, response_wait_mode: ClientToolResponseWaitMode = "timeout", interrupt_on_result: bool = False, ) -> dict[str, Any]: del timeout_seconds, wait_for_response, response_wait_mode, interrupt_on_result for index, pending in enumerate(self._results): if pending.get("name") == function_name: self._results.pop(index) return dict(pending.get("result") or {}) item = { "id": new_id("fc"), "type": "function_call", "status": "completed", "call_id": new_id("call"), "name": function_name, "arguments": json.dumps(arguments, ensure_ascii=False), } self.calls.append(item) self.called.set() # The HTTP response carries this call back to the caller. The worker is # cancelled by ResponsesTextRunner after collecting parallel calls. await asyncio.Future() raise AssertionError("unreachable") def _context_snapshot( context: LLMContext, *, function_calls: list[dict[str, Any]], fallback_messages: list[dict[str, Any]] | None = None, ) -> list[dict[str, Any]]: """Keep model-visible state and align provider call IDs with public IDs.""" messages = [ deepcopy(message) for message in context.get_messages() if message.get("role") != "system" ] if fallback_messages: messages.extend(deepcopy(fallback_messages)) for public_call in function_calls: matched = False for message in reversed(messages): for tool_call in reversed(message.get("tool_calls") or []): function = tool_call.get("function") or {} if function.get("name") != public_call["name"]: continue tool_call["id"] = public_call["call_id"] matched = True break if matched: break if not matched: messages.append( { "role": "assistant", "content": "", "tool_calls": [ { "id": public_call["call_id"], "type": "function", "function": { "name": public_call["name"], "arguments": public_call["arguments"], }, } ], } ) return json.loads(json.dumps(messages, ensure_ascii=False, default=str)) def _public_variables(brain, cfg: AssistantConfig) -> dict[str, Any]: store = getattr(brain, "_store", None) public_values = getattr(store, "public_values", None) if callable(public_values): return dict(public_values()) return { name: value for name, value in (cfg.dynamic_variables or {}).items() if not name.startswith(("system__", "secret__")) } def _runtime_state(brain) -> dict[str, Any]: snapshot = getattr(brain, "text_state_snapshot", None) return dict(snapshot()) if callable(snapshot) else {} async def _wait_until_idle( capture: _CaptureProcessor, runner_task: asyncio.Task[None], *, require_activity: bool, client_tools: ResponsesClientToolPort | None = None, ) -> str: while True: if capture.error: raise ResponseRunError(capture.error) if client_tools is not None and client_tools.called.is_set(): # Let calls emitted in the same provider batch enter the port. await asyncio.sleep(0.05) return "function_call" if runner_task.done(): await runner_task if capture.outputs or capture.deltas: return "text" raise ResponseRunError("Pipeline ended before producing a response") active = capture.window.activity_count > 0 settled = time.monotonic() - capture.window.last_activity_at >= IDLE_SETTLE_SECONDS if capture.window.idle and settled and (active or not require_activity): return "text" await asyncio.sleep(0.05) class ResponsesTextRunner: async def run( self, cfg: AssistantConfig, *, history_messages: list[dict[str, Any]], current_messages: list[dict[str, Any]], runtime_state: dict[str, Any] | None = None, client_tool_results: list[dict[str, Any]] | None = None, on_delta: DeltaCallback | None = None, ) -> ResponseRunResult: if cfg.runtimeMode != "pipeline": raise ResponseRunError("Responses API only supports pipeline assistants") if not current_messages: raise ResponseRunError("The current response has no usable input") brain = build_brain(cfg) restore_state = getattr(brain, "restore_text_state", None) if runtime_state and callable(restore_state): restore_state(runtime_state) knowledge_config = cfg.knowledge_retrieval_config or {} knowledge_mode = str(knowledge_config.get("mode") or "automatic") def with_knowledge_hint(text: str) -> str: if cfg.type != "prompt" or not cfg.knowledge_base_id: return text mode_text = ( "系统会在每轮用户问题前自动提供相关资料。" if knowledge_mode == "automatic" else "需要业务资料时先调用 search_knowledge_base。" ) return f"{text}\n\n你已连接内部知识库。{mode_text}" system_prompt = with_knowledge_hint(brain.system_prompt(cfg)) context = LLMContext( messages=( list(history_messages) if cfg.type == "workflow" else [ {"role": "system", "content": system_prompt}, *history_messages, ] ) ) graph_settings = cfg.graph.get("settings") or {} default_resource = cfg.workflow_model_resources.get( str(graph_settings.get("defaultLlmResourceId") or "") ) llm_cfg = ( config_with_resource(cfg, default_resource) if cfg.type == "workflow" and default_resource is not None else cfg ) llm = brain.build_llm(llm_cfg, context) llm_services: dict[str, FrameProcessor] = {} current_llm_service: FrameProcessor = llm if cfg.type == "workflow": llm, llm_services, current_llm_service = build_workflow_llm_switcher( cfg, llm ) aggregators = LLMContextAggregatorPair(context) user_aggregator = aggregators.user() assistant_aggregator = aggregators.assistant() capture = _CaptureProcessor(on_delta) knowledge = KnowledgeRetrievalProcessor( cfg.knowledge_base_id if knowledge_mode == "automatic" else None, top_n=int(knowledge_config.get("top_n", knowledge_config.get("topN", 5))), score_threshold=float( knowledge_config.get( "score_threshold", knowledge_config.get("scoreThreshold", 0.0) ) ), ) pipeline = Pipeline( [ user_aggregator, UserTurnRoutingProcessor(brain), knowledge, llm, capture, assistant_aggregator, ] ) worker = PipelineWorker( pipeline, params=PipelineParams(enable_metrics=False), enable_rtvi=False, enable_turn_tracking=False, idle_timeout_secs=None, ) runner = WorkerRunner(handle_sigint=False, check_dangling_tasks=False) worker_started = asyncio.Event() @worker.event_handler("on_pipeline_started") async def on_pipeline_started(_worker, _frame): worker_started.set() @worker.event_handler("on_pipeline_error") async def on_pipeline_error(_worker, frame): capture.error = frame.error or "Pipeline execution failed" capture.window.touch() client_tools = ResponsesClientToolPort(client_tool_results) workflow_vision_scope: dict[str, Any] = {"enabled": False} async def queue_call_end(reason: str) -> None: await worker.queue_frame(EndFrame(reason=reason)) call_end = CallEndCoordinator(queue_call_end) current_service = current_llm_service async def switch_services( llm_resource_id: str | None, _asr_resource_id: str | None, _tts_resource_id: str | None, ) -> None: nonlocal current_service target = ( llm_services.get(llm_resource_id) if llm_resource_id else current_llm_service ) if target is None: raise ValueError(f"Workflow LLM resource not found: {llm_resource_id}") if target is current_service: return await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) current_service = target def set_system_prompt(text: str) -> None: messages = context.get_messages() text = with_knowledge_hint(text) if messages and messages[0].get("role") == "system": messages[0] = {"role": "system", "content": text} else: messages.insert(0, {"role": "system", "content": text}) knowledge_schema: FunctionSchema | None = None if cfg.type == "prompt" and cfg.knowledge_base_id and knowledge_mode == "on_demand": knowledge_schema = FunctionSchema( name="search_knowledge_base", description="在当前助手绑定的知识库中检索资料。", properties={ "query": {"type": "string", "description": "完整问题或检索关键词"} }, required=["query"], ) async def search_bound_knowledge(params: FunctionCallParams) -> None: query = str(params.arguments.get("query") or "").strip() if not query: await params.result_callback( {"status": "error", "message": "检索问题为空"} ) return try: async with SessionLocal() as session: results = await search_knowledge( session, cfg.knowledge_base_id or "", query, top_k=int( knowledge_config.get( "top_n", knowledge_config.get("topN", 5) ) ), score_threshold=float( knowledge_config.get( "score_threshold", knowledge_config.get("scoreThreshold", 0.0), ) ), ) await params.result_callback({"status": "ok", "results": results}) except Exception as exc: # noqa: BLE001 logger.warning(f"Responses knowledge search failed: {exc}") await params.result_callback( {"status": "error", "message": "知识库检索暂时不可用"} ) llm.register_function("search_knowledge_base", search_bound_knowledge) def set_tools(schemas=None) -> None: visible = list(schemas or []) if knowledge_schema is not None: visible.append(knowledge_schema) if visible: context.set_tools(ToolsSchema(standard_tools=visible)) else: context.set_tools() await brain.setup( cfg, BrainRuntime( context=context, llm=llm, queue_frame=worker.queue_frame, set_system_prompt=set_system_prompt, set_tools=set_tools, call_end=call_end, session_id=cfg.conversation_id, client_tools=client_tools, worker=worker, context_aggregator=aggregators, transport=None, switch_services=switch_services, set_knowledge_scope=knowledge.set_scope, set_vision_scope=lambda scope: workflow_vision_scope.update(scope), vision_function=None, set_input_enabled=lambda _enabled: None, apply_turn_config=lambda _enabled, _config: asyncio.sleep(0), flow_global_functions=[], ), ) greeting = str(await brain.greeting(cfg) or "").strip() if greeting and brain.spec.owns_context: brain.prepare_greeting_context(greeting, context) @assistant_aggregator.event_handler("on_assistant_turn_started") async def on_assistant_turn_started(_aggregator): capture.window.active_assistant += 1 capture.window.touch() await brain.on_assistant_text_start(new_id("turn")) @assistant_aggregator.event_handler("on_assistant_turn_stopped") async def on_assistant_turn_stopped(_aggregator, message): text = str(message.content or "").strip() capture.window.active_assistant = max( 0, capture.window.active_assistant - 1 ) if text: capture.outputs.append(text) capture.window.touch() await brain.on_assistant_text_end( new_id("turn"), text, bool(getattr(message, "interrupted", False)), ) if call_end.ending: await call_end.finish() runner_task: asyncio.Task[None] | None = None try: await brain.run_preflight() await runner.add_workers(worker) runner_task = asyncio.create_task( runner.run(), name=f"responses-{new_id('run')[:20]}" ) await asyncio.wait_for( worker_started.wait(), timeout=WORKER_START_TIMEOUT_SECONDS ) await brain.on_connected(greeting_pending=False) await brain.on_client_ready() boundary = asyncio.Event() await worker.queue_frame(_BoundaryFrame(reached=boundary)) await asyncio.wait_for(boundary.wait(), timeout=WORKER_START_TIMEOUT_SECONDS) startup_outcome = await _wait_until_idle( capture, runner_task, require_activity=False, client_tools=client_tools, ) if startup_outcome == "function_call": startup_text = ( capture.outputs[-1] if capture.outputs else "".join(capture.deltas) ).strip() return ResponseRunResult( text=startup_text, function_calls=list(client_tools.calls), context_messages=_context_snapshot( context, function_calls=client_tools.calls, fallback_messages=current_messages, ), dynamic_variables=_public_variables(brain, cfg), runtime_state=_runtime_state(brain), ) capture.reset() await worker.queue_frame( LLMMessagesAppendFrame(messages=current_messages, run_llm=True) ) outcome = await _wait_until_idle( capture, runner_task, require_activity=True, client_tools=client_tools, ) if outcome == "function_call": text = ( capture.outputs[-1] if capture.outputs else "".join(capture.deltas) ).strip() return ResponseRunResult( text=text, function_calls=list(client_tools.calls), context_messages=_context_snapshot( context, function_calls=client_tools.calls, ), dynamic_variables=_public_variables(brain, cfg), runtime_state=_runtime_state(brain), ) text = (capture.outputs[-1] if capture.outputs else "".join(capture.deltas)).strip() if not text: raise ResponseRunError("Model returned no text") return ResponseRunResult( text=text, function_calls=[], context_messages=_context_snapshot(context, function_calls=[]), dynamic_variables=_public_variables(brain, cfg), runtime_state=_runtime_state(brain), ) finally: if not worker.has_finished(): await worker.cancel(reason="responses_turn_complete") if runner_task is not None: await asyncio.gather(runner_task, return_exceptions=True)