308 lines
16 KiB
Python
308 lines
16 KiB
Python
"""Task-local Reasoner built on Pipecat's public backend worker contract.
|
|
|
|
Pipecat owns the context, model/tool loop and Jobs. This adapter only turns
|
|
explicit task controls into the same persisted events as a remote backend.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Awaitable, Callable
|
|
from copy import deepcopy
|
|
from uuid import uuid4
|
|
|
|
from models import AssistantConfig, RuntimeModelResource, RuntimeTaskBackend
|
|
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
|
from pipecat.pipeline.job_context import JobEvent, JobParams
|
|
from pipecat.processors.aggregators.llm_context import LLMContext
|
|
from pipecat.services.llm_service import FunctionCallParams, FunctionCallResultProperties
|
|
from pipecat.services.settings import LLMSettings
|
|
from pipecat.workers.llm.backend_llm_worker import (
|
|
ATTACHED_UPDATE_TYPE,
|
|
ATTACH_JOB_NAME,
|
|
ERROR_UPDATE_TYPE,
|
|
IDLE_UPDATE_TYPE,
|
|
MESSAGE_JOB_NAME,
|
|
TOOL_CALL_UPDATE_TYPE,
|
|
BackendLLMWorker,
|
|
)
|
|
|
|
from services.background_tasks.native import NativeTaskRuntime
|
|
from services.pipecat.service_factory import config_with_resource, create_llm
|
|
from services.runtime_variables import DynamicVariableStore
|
|
from services.system_tools import RESERVED_TOOL_NAMES
|
|
from services.tool_executor import ToolExecutionError, ToolExecutor
|
|
|
|
Emit = Callable[[dict], Awaitable[None]]
|
|
CONTROL_NAMES = frozenset({"complete_task", "ask_user"})
|
|
|
|
TASK_INSTRUCTIONS = """你是语音助手的后台任务执行者，有独立上下文。
|
|
只执行委派目标，并使用当前已授权的工具；技能说明不能增加权限。
|
|
结束执行时必须调用 complete_task(status, result)，明确报告任务目标是否完成。
|
|
只有完整完成目标时 status 才为 succeeded；无法完成或只完成部分目标时必须使用 failed，
|
|
并在 result 中说明已完成内容、未完成部分和限制。该状态是你的执行报告，不代表平台独立验证。
|
|
需要用户提供必要信息时调用 ask_user(question)，等待回复后继续。
|
|
普通文字、思考过程、工具回执和模型空闲都不代表任务完成。
|
|
任务结果由 Speaker 整理后告知用户，不要输出内部推理过程。
|
|
"""
|
|
|
|
|
|
def create_reasoner_llm(backend: RuntimeTaskBackend):
|
|
"""Use the shared model factory, with independent settings for this task."""
|
|
resource = backend.model
|
|
if resource is None or resource.capability != "LLM":
|
|
raise ValueError("内置 Reasoner 需要配置独立的 LLM 模型资源")
|
|
effort = backend.config.get("reasoning_effort", "provider_default")
|
|
if effort not in {"provider_default", "low", "medium", "high"}:
|
|
raise ValueError("不支持的推理强度")
|
|
if effort != "provider_default" and resource.values.get("supportsReasoningEffort") is not True:
|
|
raise ValueError("该模型未声明支持 reasoning_effort，不能设置推理强度")
|
|
|
|
# Do not mutate the reusable model resource or its Speaker settings.
|
|
resource = RuntimeModelResource.model_validate(resource.model_dump(mode="json"))
|
|
if effort != "provider_default":
|
|
extra_body = resource.values.get("extraBody")
|
|
if isinstance(extra_body, dict):
|
|
# The explicit task setting wins if a model preset contains it too.
|
|
resource.values["extraBody"] = {key: value for key, value in extra_body.items() if key != "reasoning_effort"}
|
|
llm = create_llm(config_with_resource(AssistantConfig(), resource))
|
|
if effort != "provider_default":
|
|
llm.settings.apply_update(LLMSettings(extra={**llm.settings.extra, "reasoning_effort": effort}))
|
|
return llm
|
|
|
|
|
|
class _Finished(Exception):
|
|
"""Leave the long-lived attach Job through its native cancellation path."""
|
|
|
|
def __init__(self, result: dict):
|
|
super().__init__("task finished")
|
|
self.result = result
|
|
|
|
|
|
class ReasonerExecutor:
|
|
def __init__(self, config: dict, secrets: dict, *, runtime: NativeTaskRuntime):
|
|
self.config = config
|
|
self.runtime = runtime
|
|
self.backend = RuntimeTaskBackend.model_validate(config["_backend"])
|
|
self._worker: BackendLLMWorker | None = None
|
|
self._emit: Emit | None = None
|
|
self._pending: dict = {}
|
|
self._final: dict | None = None
|
|
self._confirmed: dict | None = None
|
|
self._registered: list[str] = []
|
|
self._stopped = False
|
|
self._sending_input = False
|
|
self._cleanup_lock = asyncio.Lock()
|
|
self._tools = ToolExecutor(DynamicVariableStore.from_config(AssistantConfig(
|
|
dynamic_variables=deepcopy(config.get("_variables") or {}),
|
|
dynamic_variable_definitions=deepcopy(config.get("_variable_definitions") or {}),
|
|
secret_dynamic_variables=deepcopy(config.get("_secret_variables") or {}),
|
|
)))
|
|
|
|
async def prepare(self) -> dict:
|
|
return {"kind": "reasoner", "task_id": self.config["_task_id"]}
|
|
|
|
def _system_prompt(self) -> str:
|
|
parts = [
|
|
TASK_INSTRUCTIONS,
|
|
str(self.backend.config.get("instructions") or "").strip(),
|
|
self._tools.store.render(self.backend.instructions).strip(),
|
|
]
|
|
for skill in self.backend.config.get("skills") or []:
|
|
parts.append(f"技能：{skill['name']}\n{skill['content']}")
|
|
return "\n\n".join(part for part in parts if part)
|
|
|
|
def _build_worker(self) -> BackendLLMWorker:
|
|
bindings = {binding.tool_id: binding for binding in self.backend.tool_bindings}
|
|
names = set(CONTROL_NAMES | RESERVED_TOOL_NAMES | {"delegate_task"})
|
|
schemas = []
|
|
callbacks = []
|
|
for tool in self.backend.tools:
|
|
if tool.type not in {"http", "mcp"}:
|
|
raise ValueError("Reasoner 仅允许 HTTP/MCP 服务端工具")
|
|
if tool.function_name in names:
|
|
raise ValueError("Reasoner 工具函数名重复或与任务控制函数冲突")
|
|
names.add(tool.function_name)
|
|
binding = bindings.get(tool.id)
|
|
if binding is None:
|
|
raise ValueError("Reasoner 工具必须显式绑定才可执行")
|
|
properties, required = self._tools.schema_parts(tool, binding)
|
|
schemas.append(FunctionSchema(name=tool.function_name, description=tool.description or tool.name,
|
|
properties=properties, required=required))
|
|
|
|
async def execute(params, current_tool=tool, current_binding=binding):
|
|
if self._pending or self._final or self._stopped:
|
|
await params.result_callback({"status": "error", "message": "任务已暂停或结束，不执行新工具"},
|
|
properties=FunctionCallResultProperties(run_llm=False))
|
|
return
|
|
try:
|
|
result = await self._tools.execute(current_tool, dict(params.arguments or {}), binding=current_binding)
|
|
except (ToolExecutionError, ValueError):
|
|
# Tool errors can carry URLs or sensitive payloads. Let the
|
|
# model report a useful failure without exposing those.
|
|
result = {"status": "error", "message": "服务端工具执行失败，请检查工具配置或稍后重试"}
|
|
await params.result_callback(result, properties=FunctionCallResultProperties(
|
|
run_llm=not (self._pending or self._final or self._stopped),
|
|
))
|
|
|
|
callbacks.append((tool.function_name, execute))
|
|
|
|
schemas.extend([
|
|
FunctionSchema(
|
|
name="complete_task",
|
|
description="结束执行，报告任务目标是否完整完成。部分完成或无法完成必须报告 failed。",
|
|
properties={
|
|
"status": {
|
|
"type": "string",
|
|
"enum": ["succeeded", "failed"],
|
|
"description": "执行模型对目标完成情况的报告，不能仅因执行结束而报告成功。",
|
|
},
|
|
"result": {"type": "string", "description": "执行结果、依据和未完成的限制。"},
|
|
},
|
|
required=["status", "result"],
|
|
),
|
|
FunctionSchema(name="ask_user", description="需要必要信息时暂停任务，向用户提出一个问题。",
|
|
properties={"question": {"type": "string", "description": "需要用户回答的问题。"}}, required=["question"]),
|
|
])
|
|
llm = create_reasoner_llm(self.backend)
|
|
for name, callback in callbacks + [("complete_task", self._complete_task), ("ask_user", self._ask_user)]:
|
|
# Synchronous native tools drive BackendLLMWorker's existing loop.
|
|
llm.register_function(name, callback, cancel_on_interruption=True)
|
|
self._registered.append(name)
|
|
return BackendLLMWorker(
|
|
name="reasoner-" + uuid4().hex,
|
|
llm=llm,
|
|
context=LLMContext([{"role": "system", "content": self._system_prompt()}], tools=schemas),
|
|
pairing_instruction="任务由 Speaker 发送给你。使用任务控制函数提交结果或提问；普通输出不作为给用户的通知。",
|
|
)
|
|
|
|
async def _complete_task(self, params: FunctionCallParams):
|
|
status = params.arguments.get("status")
|
|
result = str(params.arguments.get("result") or "").strip()
|
|
if status not in ("succeeded", "failed") or not result or self._pending or self._stopped:
|
|
await params.result_callback(
|
|
{"status": "error", "message": "必须提供 succeeded 或 failed 状态及非空结果，且任务不能处于等待输入或停止状态"},
|
|
properties=FunctionCallResultProperties(run_llm=not (self._pending or self._stopped)),
|
|
)
|
|
return
|
|
self._final = {
|
|
"status": status,
|
|
"summary": "Reasoner 报告任务已完成" if status == "succeeded" else "Reasoner 报告任务未完成",
|
|
"result": result,
|
|
}
|
|
await params.result_callback({"status": "accepted"}, properties=FunctionCallResultProperties(run_llm=False))
|
|
|
|
async def _ask_user(self, params: FunctionCallParams):
|
|
question = str(params.arguments.get("question") or "").strip()
|
|
if not question or self._pending or self._final or self._stopped:
|
|
await params.result_callback(
|
|
{"status": "error", "message": "请在任务运行时提出一个非空问题"},
|
|
properties=FunctionCallResultProperties(run_llm=not (self._pending or self._final or self._stopped)),
|
|
)
|
|
return
|
|
self._pending = {"id": "question_" + uuid4().hex, "kind": "question",
|
|
"questions": [{"header": "补充信息", "question": question, "options": [], "custom": True}]}
|
|
await self._emit({"status": "waiting_input", "summary": "任务需要补充信息", "input_request": deepcopy(self._pending)})
|
|
await params.result_callback({"status": "waiting_input"}, properties=FunctionCallResultProperties(run_llm=False))
|
|
|
|
async def _message(self, text: str):
|
|
if self._worker is None or self._stopped:
|
|
raise ValueError("内置 Reasoner 已停止，不能继续当前任务")
|
|
async with self.runtime.coordinator.job(self._worker.name, params=JobParams(
|
|
name=MESSAGE_JOB_NAME, payload={"request": text}, timeout=15,
|
|
)):
|
|
pass
|
|
|
|
async def run(self, handle: dict, instruction: str, emit: Emit, *, resume=False) -> dict:
|
|
if resume:
|
|
return {"status": "failed", "summary": "内置任务执行曾中断，不会自动重放工具", "result": "请核查已有操作后重新创建任务。"}
|
|
if self.runtime.runner is None:
|
|
raise ValueError("后台任务运行器尚未启动")
|
|
self._emit = emit
|
|
self._worker = self._build_worker()
|
|
try:
|
|
await self.runtime.runner.add_workers(self._worker)
|
|
async with self.runtime.coordinator.job(self._worker.name, params=JobParams(name=ATTACH_JOB_NAME)) as attached:
|
|
async for event in attached:
|
|
if event.type != JobEvent.UPDATE:
|
|
continue
|
|
data = event.data or {}
|
|
kind = data.get("type")
|
|
if kind == ATTACHED_UPDATE_TYPE:
|
|
await emit({"status": "running", "summary": "内置 Reasoner 正在执行任务"})
|
|
await self._message(instruction)
|
|
elif kind == TOOL_CALL_UPDATE_TYPE and data.get("function_name") not in CONTROL_NAMES:
|
|
if not self._pending:
|
|
await emit({"status": "running", "summary": f"服务端工具 · {data.get('function_name', '')}"[:300]})
|
|
elif kind == ERROR_UPDATE_TYPE:
|
|
raise _Finished({"status": "failed", "summary": "内置推理服务执行失败", "result": "请检查模型连接和工具配置。"})
|
|
elif kind == IDLE_UPDATE_TYPE:
|
|
# An old idle update can already be on the bus when a
|
|
# user reply starts the next native message Job.
|
|
if self._sending_input or self._worker.working:
|
|
continue
|
|
if self._final:
|
|
raise _Finished(self._final)
|
|
if not self._pending:
|
|
raise _Finished({"status": "failed", "summary": "模型没有明确完成任务", "result": "模型已停止生成，但没有调用 complete_task 或 ask_user；请调整任务指令或模型后重试。"})
|
|
return {"status": "failed", "summary": "内置任务连接已中断", "result": "不会自动重放工具调用。"}
|
|
except _Finished as finished:
|
|
self._confirmed = finished.result
|
|
return finished.result
|
|
finally:
|
|
await self._cleanup()
|
|
|
|
async def _cleanup(self):
|
|
async with self._cleanup_lock:
|
|
worker = self._worker
|
|
self._stopped = True
|
|
if worker is None:
|
|
return
|
|
# This is PipelineWorker.cancel(), which cancels only this pipeline.
|
|
# BaseWorker.cancel()/runner.cancel() would stop unrelated tasks too.
|
|
await worker.cancel(reason="task finished or cancelled")
|
|
await asyncio.wait_for(worker.wait(), 10)
|
|
# WorkerRunner retains ended worker registrations. Release transcript
|
|
# and handler captures through public APIs instead of editing Pipecat.
|
|
worker.context.set_messages([])
|
|
worker.context.set_tools([])
|
|
for name in self._registered:
|
|
worker.llm.unregister_function(name)
|
|
self._registered.clear()
|
|
self._worker = None
|
|
|
|
async def cancel(self, handle: dict) -> bool:
|
|
if self._worker is not None:
|
|
await self._cleanup()
|
|
return True
|
|
|
|
async def snapshot(self, handle: dict) -> dict:
|
|
if self._confirmed and self._worker is None:
|
|
return dict(self._confirmed)
|
|
if self._pending:
|
|
return {"status": "waiting_input", "summary": "任务需要补充信息", "input_request": deepcopy(self._pending)}
|
|
if self._stopped:
|
|
return {"status": "failed", "summary": "内置任务已中断", "result": "不会自动重放工具调用。"}
|
|
return {"status": "running", "summary": "内置 Reasoner 正在执行任务"}
|
|
|
|
async def send_input(self, request: dict, value: dict):
|
|
if not self._pending or request.get("id") != self._pending["id"]:
|
|
raise ValueError("该问题已处理或不属于当前任务")
|
|
answers = value.get("answers")
|
|
if not isinstance(answers, list) or len(answers) != 1 or not isinstance(answers[0], list):
|
|
raise ValueError("请回答当前任务的问题")
|
|
text = "\n".join(str(answer).strip() for answer in answers[0]).strip()
|
|
if not text:
|
|
raise ValueError("回答不能为空")
|
|
previous = self._pending
|
|
self._pending = {}
|
|
self._sending_input = True
|
|
try:
|
|
await self._message("用户补充信息：\n" + text)
|
|
except Exception:
|
|
if not self._pending and not self._stopped:
|
|
self._pending = previous
|
|
raise
|
|
finally:
|
|
self._sending_input = False
|