wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance workflow engine and integration in backend and frontend

Browse Source 
- Introduce a new WorkflowEngine class to manage workflow graphs, enabling dynamic node-based interactions.
- Update AssistantConfig to include a graph field for workflow definitions, allowing for flexible configuration.
- Modify pipeline execution to support workflow-driven dialogue, integrating node transitions and system prompts based on active nodes.
- Enhance frontend components to visualize active nodes and provide debugging capabilities, including highlighting the current node during interactions.
- Refactor existing components to accommodate new workflow functionalities and improve overall user experience.
This commit is contained in:
Xin Wang
2026-06-15 15:32:10 +08:00
parent c2a39257ff
commit aae0342a57
10 changed files with 361 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

153
backend/services/pipecat/pipeline.py
View File

@@ -8,10 +8,14 @@
from uuid import uuid4
import config
from loguru import logger
from models import AssistantConfig
from services.pipecat.service_factory import create_realtime_service, create_services
from services.workflow_engine import WorkflowEngine
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import (
EndFrame,
@@ -209,7 +213,32 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
stt, llm, tts = create_services(cfg)
context = LLMContext(messages=[{"role": "system", "content": cfg.prompt}])
# ---- workflow 图引擎(可选)----
# 有节点图时按图驱动:开场白/系统提示来自起始节点,每轮回复后按条件路由。
engine = WorkflowEngine(cfg.graph or {})
workflow_active = engine.has_graph()
wf_state = {
# 开始节点本身就是会话节点(有自己的 prompt,可多轮),从它开始
"current": engine.start_id if workflow_active else None,
"ended": False,
"turns_in_node": 0,
"end_frame_queued": False,
}
history: list[dict] = []
# 当前节点没有可调用转移工具(全是空条件)时,才启用文本兜底路由
FALLBACK_AFTER_TURNS = 2
if workflow_active:
greeting = engine.greeting() or cfg.greeting
system_content = engine.system_prompt_for(wf_state["current"])
logger.info(
f"工作流模式启用: 起始节点={engine.name(wf_state['current'])}"
)
else:
greeting = cfg.greeting
system_content = cfg.prompt
context = LLMContext(messages=[{"role": "system", "content": system_content}])
user_aggregator = LLMUserAggregator(
context,
params=LLMUserAggregatorParams(
@@ -267,6 +296,96 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
greeting_transcript_sent = False
pending_text_inputs: list[str] = []
async def emit_node_active(node_id: str | None) -> None:
"""通知前端当前激活的节点,画布据此高亮。"""
if node_id:
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message={"type": "node-active", "nodeId": node_id}
)
)
def set_system_prompt(text: str) -> None:
"""替换上下文里的系统提示(节点切换时整体替换,而非追加)。"""
messages = context.get_messages()
if messages and messages[0].get("role") == "system":
messages[0] = {"role": "system", "content": text}
else:
messages.insert(0, {"role": "system", "content": text})
def apply_node(node_id: str | None) -> None:
"""进入节点:设置系统提示 + 把出边注册为可调用的转移工具。"""
set_system_prompt(engine.system_prompt_for(node_id))
if engine.is_end(node_id):
context.set_tools() # 终止节点无工具
return
schemas = [
FunctionSchema(
name=engine.edge_fn_name(edge),
description=engine.edge_description(edge),
properties={},
required=[],
)
for edge in engine.outgoing(node_id)
]
if schemas:
context.set_tools(ToolsSchema(standard_tools=schemas))
else:
context.set_tools() # 无出边:清空工具
async def go_to_node(target: str) -> None:
"""执行转移:切当前节点、重置计数、点亮画布、设置提示/工具。
结束节点:设 ended 标记,apply_node 会清空工具,模型据结束语提示说完后,
on_assistant_text_end 里排入 EndFrame 挂断,不再多轮。
"""
wf_state["current"] = target
wf_state["turns_in_node"] = 0
if engine.is_end(target):
wf_state["ended"] = True
await emit_node_active(target)
apply_node(target)
def make_transition_handler(target: str):
async def handler(params):
logger.info(f"LLM 触发转移 → {engine.name(target)}")
await go_to_node(target)
# 返回工具结果,pipecat 随即在新节点的提示/工具下继续生成
await params.result_callback({"status": "ok"})
return handler
async def fallback_route() -> None:
"""文本兜底:模型迟迟不调用转移工具时,用一次轻量分类器判断是否转移。"""
if not workflow_active or wf_state["ended"]:
return
if wf_state["turns_in_node"] < FALLBACK_AFTER_TURNS:
return
if not engine.outgoing(wf_state["current"]):
return
target = await engine.route(
wf_state["current"],
history,
api_key=cfg.llm_api_key or config.LLM_API_KEY,
base_url=cfg.llm_base_url or config.LLM_BASE_URL,
model=cfg.model or config.LLM_MODEL,
)
if target and target != wf_state["current"]:
logger.info(f"文本兜底触发转移 → {engine.name(target)}")
# 仅切换节点提示/工具,下一轮用户输入即在新节点处理
await go_to_node(target)
# 把每条边注册成 LLM 可调用的转移函数(按边唯一命名,处理器全局注册一次,
# 由各节点的 context.tools 控制当前可见哪些)。
if workflow_active:
for edge in engine.edges:
target = edge.get("target")
if target:
llm.register_function(
engine.edge_fn_name(edge), make_transition_handler(target)
)
apply_node(wf_state["current"]) # 设初始节点的提示与工具
async def append_user_text_to_context(text: str, *, run_llm: bool) -> None:
await worker.queue_frame(
LLMMessagesAppendFrame(
@@ -277,6 +396,8 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(_aggregator, _strategy, message):
if message.content:
history.append({"role": "user", "content": message.content})
await queue_transcript("user", message.content, message.timestamp)
@assistant_aggregator.event_handler("on_assistant_text_start")
@@ -315,10 +436,26 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
}
)
)
# 助手把话说完(未被打断)后:累加本节点轮次,必要时走文本兜底路由。
# 正常情况下转移由 LLM 直接调用转移工具完成(go_to_node),无需这里处理。
if content and not interrupted and workflow_active:
history.append({"role": "assistant", "content": content})
if wf_state["ended"]:
# 结束节点:说完结束语后挂断,不再继续多轮对话
if not wf_state["end_frame_queued"]:
wf_state["end_frame_queued"] = True
logger.info("结束节点结束语已播报,挂断通话")
await worker.queue_frame(EndFrame())
else:
wf_state["turns_in_node"] += 1
await fallback_route()
elif content and not interrupted:
history.append({"role": "assistant", "content": content})
@text_input.event_handler("on_text_input")
async def on_text_input(_processor, text):
pending_text_inputs.append(text)
history.append({"role": "user", "content": text})
# 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。
await queue_transcript("user", text, time_now_iso8601())
@@ -333,21 +470,25 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
@text_input.event_handler("on_text_append")
async def on_text_append(_processor, text):
# 静默追加:写进上下文但不打断、不触发推理;transcript 照常上报
history.append({"role": "user", "content": text})
await queue_transcript("user", text, time_now_iso8601())
await append_user_text_to_context(text, run_llm=False)
@text_input.event_handler("on_client_ready")
async def on_client_ready(_processor):
nonlocal greeting_transcript_sent
if cfg.greeting and not greeting_transcript_sent:
if greeting and not greeting_transcript_sent:
greeting_transcript_sent = True
await queue_transcript("assistant", cfg.greeting, time_now_iso8601())
await queue_transcript("assistant", greeting, time_now_iso8601())
@transport.event_handler("on_client_connected")
async def on_client_connected(_transport, _client):
if cfg.greeting:
context.add_message({"role": "assistant", "content": cfg.greeting})
await worker.queue_frame(TTSSpeakFrame(cfg.greeting, append_to_context=False))
if greeting:
context.add_message({"role": "assistant", "content": greeting})
await worker.queue_frame(TTSSpeakFrame(greeting, append_to_context=False))
# 工作流:点亮当前(开始)节点。开始节点即首个会话节点。
if workflow_active:
await emit_node_active(wf_state["current"])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(_transport, _client):