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

Add Dify integration and enhance workflow node specifications

Browse Source 
- Introduce new fields `dify_api_url` and `dify_api_key` in `AssistantConfig` for Dify API integration.
- Update `requirements.txt` to include `dify-client-python` for Dify SDK support.
- Modify `config_resolver` to handle Dify connection information.
- Add a new `globalNode` type in workflow specifications to provide unified settings across workflows.
- Enhance node specifications with additional constraints and default values for better configuration management.
- Update frontend components to support the new `globalNode` type and its properties, improving workflow editor functionality.
This commit is contained in:
Xin Wang
2026-07-11 22:26:31 +08:00
parent dfb9c5bd11
commit 00270a5c01
23 changed files with 1270 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
backend/services/pipecat/call_lifecycle.py Normal file
View File

@@ -0,0 +1,60 @@
"""Shared call termination timing for prompt tools and workflow end nodes."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from loguru import logger
from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class CallEndCoordinator:
"""End immediately or after the currently armed closing speech finishes."""
def __init__(self, queue_end: Callable[[str], Awaitable[None]]):
self._queue_end = queue_end
self._ending = False
self._armed = False
self._speaking = False
self._finished = False
self._reason = "completed"
@property
def ending(self) -> bool:
return self._ending
def begin(self, reason: str) -> None:
self._ending = True
self._reason = reason or "completed"
def arm_after_speech(self) -> None:
self._armed = True
async def finish(self) -> None:
if self._finished:
return
self._finished = True
await self._queue_end(self._reason)
async def observe(self, frame) -> None:
if isinstance(frame, BotStartedSpeakingFrame) and self._armed:
self._speaking = True
elif (
isinstance(frame, BotStoppedSpeakingFrame)
and self._armed
and self._speaking
):
logger.info("结束语播报完毕,挂断通话")
await self.finish()
class EndCallAfterSpeechProcessor(FrameProcessor):
def __init__(self, coordinator: CallEndCoordinator):
super().__init__()
self._coordinator = coordinator
async def process_frame(self, frame, direction: FrameDirection):
await super().process_frame(frame, direction)
await self.push_frame(frame, direction)
await self._coordinator.observe(frame)

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

@@ -3,7 +3,7 @@
关键设计:**transport 由调用方传入**,管线本身不关心是 WebRTC 还是 WS。
这就是"同时支持多种输出"的落点——加输出方式不用动这里。
对应 dograh 的 pipeline_builder.py + run_pipeline.py(已砍掉 workflow 引擎/DB/录音/指标)。
对话编排交给 Brain;本文件只保留共享媒体管线、输入输出和通话生命周期。
"""
import asyncio
@@ -16,21 +16,22 @@ from loguru import logger
from models import AssistantConfig
from openai import AsyncOpenAI
from PIL import Image
from services.brains import build_brain
from services.brains import Brain, BrainRuntime, build_brain
from services.conversation_history import ConversationRecorder
from services.pipecat.call_lifecycle import (
CallEndCoordinator,
EndCallAfterSpeechProcessor,
)
from services.pipecat.service_factory import (
create_realtime_service,
create_stt,
create_tts,
)
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 (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
EndFrame,
InputTransportMessageFrame,
InterruptionFrame,
@@ -57,10 +58,7 @@ from pipecat.runner.utils import (
get_transport_client_id,
maybe_capture_participant_camera,
)
from pipecat.services.llm_service import (
FunctionCallParams,
FunctionCallResultProperties,
)
from pipecat.services.llm_service import FunctionCallParams
from pipecat.turns.user_start import (
TranscriptionUserTurnStartStrategy,
VADUserTurnStartStrategy,
@@ -399,8 +397,7 @@ async def run_pipeline(
cfg.runtimeMode == "realtime"
and "realtime" not in brain.spec.supported_runtime_modes
):
logger.warning(f"类型 {cfg.type} 不支持 realtime,回退 cascade")
cfg.runtimeMode = "pipeline"
raise ValueError(f"类型 {cfg.type} 不支持 realtime 运行模式")
if cfg.runtimeMode == "realtime":
if vision_enabled:
@@ -408,6 +405,7 @@ async def run_pipeline(
await run_realtime_pipeline(
transport,
cfg,
brain=brain,
assistant_id=assistant_id,
channel=channel,
)
@@ -416,45 +414,24 @@ async def run_pipeline(
stt = create_stt(cfg)
tts = create_tts(cfg)
# ---- 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_turn_id": None, # 结束节点回复的 turn_id(其 text_start 在 ended 之后)
"end_armed": False, # 结束语文本已生成完(已下发 data channel)
"end_speaking": False, # 结束语音频已开始播报
"end_frame_queued": False,
}
call_end_state = {
"ending": False,
"armed": False,
"speaking": False,
"frame_queued": False,
"reason": "completed",
}
history: list[dict] = []
# 当前节点没有可调用转移工具(全是空条件)时,才启用文本兜底路由
FALLBACK_AFTER_TURNS = 2
greeting = await brain.greeting(cfg)
system_content = brain.system_prompt(cfg)
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'])}"
worker_holder: dict = {}
async def queue_call_end(reason: str) -> None:
worker = worker_holder.get("worker")
if worker is None:
return
logger.info(f"结束通话: reason={reason}")
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message={"type": "call-ended", "reason": reason}
)
)
elif brain.spec.owns_context:
greeting = cfg.greeting
system_content = cfg.prompt
else:
# 外部托管(fastgpt 等):开场白来自对方后台,系统提示/上下文不归我们维护
greeting = await brain.greeting(cfg)
system_content = ""
await worker.queue_frame(EndFrame())
call_end = CallEndCoordinator(queue_call_end)
def with_vision_hint(text: str) -> str:
if not vision_enabled:
@@ -466,7 +443,7 @@ async def run_pipeline(
context = LLMContext(
messages=[{"role": "system", "content": with_vision_hint(system_content)}]
)
# LLM 槽由大脑提供:内部类型=OpenAI 兼容服务;fastgpt=包 SDK 的伪 LLM。
# LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。
llm = brain.build_llm(cfg, context)
user_aggregator = LLMUserAggregator(
context,
@@ -474,9 +451,7 @@ async def run_pipeline(
vad_analyzer=SileroVADAnalyzer(),
user_mute_strategies=[
FunctionCallUserMuteStrategy(),
CallEndingUserMuteStrategy(
lambda: bool(call_end_state["ending"])
),
CallEndingUserMuteStrategy(lambda: call_end.ending),
],
user_turn_strategies=UserTurnStrategies(
start=[
@@ -489,9 +464,7 @@ async def run_pipeline(
),
)
assistant_aggregator = PassthroughLLMAssistantAggregator(context)
text_input = TextInputProcessor(
should_ignore_input=lambda: bool(call_end_state["ending"])
)
text_input = TextInputProcessor(should_ignore_input=lambda: call_end.ending)
vision_capture = VisionCaptureProcessor()
vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg)
vision_state: dict[str, str | None] = {"client_id": None}
@@ -572,99 +545,6 @@ async def run_pipeline(
if vision_enabled:
llm.register_function(VISION_TOOL_NAME, fetch_user_image)
end_call_tools = [
tool
for tool in cfg.tools
if cfg.type == "prompt" and tool.type == "end_call"
]
end_call_schemas: list[FunctionSchema] = []
worker_holder: dict = {}
async def queue_call_end(reason: str) -> None:
if call_end_state["frame_queued"] or worker_holder.get("worker") is None:
return
call_end_state["frame_queued"] = True
logger.info(f"结束通话: reason={reason}")
await worker_holder["worker"].queue_frame(
OutputTransportMessageUrgentFrame(
message={"type": "call-ended", "reason": reason}
)
)
await worker_holder["worker"].queue_frame(EndFrame())
def make_end_call_handler(tool):
config = (tool.definition or {}).get("config") or {}
message_type = str(config.get("message_type") or "none")
custom_message = str(config.get("custom_message") or "").strip()
capture_reason = bool(config.get("capture_reason", True))
async def end_call(params: FunctionCallParams) -> None:
reason = str(params.arguments.get("reason") or "end_call_tool").strip()
call_end_state["ending"] = True
logger.info(
f"End Call Tool EXECUTED: {tool.function_name}, reason={reason}"
)
await params.result_callback(
{"status": "success", "action": "ending_call"},
properties=FunctionCallResultProperties(run_llm=False),
)
if message_type != "custom" or not custom_message:
await queue_call_end(reason)
return
call_end_state["reason"] = reason
call_end_state["armed"] = True
turn_id = uuid4().hex
timestamp = time_now_iso8601()
for message in (
{
"type": "assistant-text-start",
"turn_id": turn_id,
"timestamp": timestamp,
},
{
"type": "assistant-text-delta",
"turn_id": turn_id,
"delta": custom_message,
},
{
"type": "assistant-text-end",
"turn_id": turn_id,
"content": custom_message,
"interrupted": False,
},
):
await worker_holder["worker"].queue_frame(
OutputTransportMessageUrgentFrame(message=message)
)
await worker_holder["worker"].queue_frame(
TTSSpeakFrame(custom_message, append_to_context=False)
)
properties = (
{
"reason": {
"type": "string",
"description": "结束本次通话的简短原因。",
}
}
if capture_reason
else {}
)
schema = FunctionSchema(
name=tool.function_name,
description=tool.description or "结束当前通话。",
properties=properties,
required=["reason"] if capture_reason else [],
)
return schema, end_call
for end_call_tool in end_call_tools:
schema, handler = make_end_call_handler(end_call_tool)
end_call_schemas.append(schema)
llm.register_function(end_call_tool.function_name, handler)
def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None:
tools = list(schemas or [])
if vision_enabled:
@@ -674,31 +554,6 @@ async def run_pipeline(
else:
context.set_tools()
# Workflow 结束节点和 end_call 固定结束语都等到 BotStoppedSpeakingFrame
# 再挂断,确保文字(data channel)与音频完整送达。
class EndCallAfterSpeech(FrameProcessor):
async def process_frame(self, frame, direction: FrameDirection):
await super().process_frame(frame, direction)
await self.push_frame(frame, direction)
armed = wf_state["end_armed"] or call_end_state["armed"]
# 结束语文本生成完(end_armed)→ 其音频开始(end_speaking)→ 音频说完才挂断。
# 配对 started/stopped,避免被结束节点之前的话(如先答一句再转移)的
# stopped 事件提前触发,导致结束语被截断。
if isinstance(frame, BotStartedSpeakingFrame) and armed:
if wf_state["end_armed"]:
wf_state["end_speaking"] = True
if call_end_state["armed"]:
call_end_state["speaking"] = True
elif (
isinstance(frame, BotStoppedSpeakingFrame)
and (wf_state["end_speaking"] or call_end_state["speaking"])
and worker_holder.get("worker") is not None
):
logger.info("结束语播报完毕,挂断通话")
wf_state["end_frame_queued"] = True
reason = str(call_end_state["reason"] or "completed")
await queue_call_end(reason)
recorder = await ConversationRecorder.start(
assistant_id=assistant_id,
assistant_name=cfg.name,
@@ -718,7 +573,7 @@ async def run_pipeline(
# waiting for a TTS provider to emit spoken-text/timestamp frames.
assistant_aggregator,
tts,
EndCallAfterSpeech(),
EndCallAfterSpeechProcessor(call_end),
ConversationHistoryProcessor(recorder),
transport.output(),
]
@@ -749,15 +604,6 @@ async def run_pipeline(
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()
@@ -767,89 +613,18 @@ async def run_pipeline(
else:
messages.insert(0, {"role": "system", "content": content})
def apply_node(node_id: str | None) -> None:
"""进入节点:设置系统提示 + 把出边注册为可调用的转移工具。"""
set_system_prompt(engine.system_prompt_for(node_id))
if engine.is_end(node_id):
set_visible_tools([]) # 终止节点不展示转移工具,但保留视觉工具
return
schemas = [
FunctionSchema(
name=engine.edge_fn_name(edge),
description=engine.edge_description(edge),
properties={},
required=[],
)
for edge in engine.outgoing(node_id)
]
set_visible_tools(schemas)
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)
async def speak_transition(edge: dict | None) -> None:
"""切换瞬间播报过渡语(可选),掩盖切节点/新一轮生成的延迟。不写入上下文。"""
speech = engine.edge_transition_speech(edge)
if speech:
await worker.queue_frame(TTSSpeakFrame(speech, append_to_context=False))
def make_transition_handler(edge: dict):
target = edge.get("target")
async def handler(params):
logger.info(f"LLM 触发转移 → {engine.name(target)}")
# 进结束节点不播过渡语(结束语本身就是收尾,避免打断挂断时序)
if not engine.is_end(target):
await speak_transition(edge)
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=_require(cfg.llm_api_key, "LLM apiKey"),
base_url=_require(cfg.llm_base_url, "LLM apiUrl"),
model=_require(cfg.model, "LLM modelId"),
)
if target and target != wf_state["current"]:
logger.info(f"文本兜底触发转移 → {engine.name(target)}")
if not engine.is_end(target):
await speak_transition(engine.find_edge(wf_state["current"], target))
# 仅切换节点提示/工具,下一轮用户输入即在新节点处理
await go_to_node(target)
# 把每条边注册成 LLM 可调用的转移函数(按边唯一命名,处理器全局注册一次,
# 由各节点的 context.tools 控制当前可见哪些)。
if workflow_active:
for edge in engine.edges:
if edge.get("target"):
llm.register_function(
engine.edge_fn_name(edge), make_transition_handler(edge)
)
apply_node(wf_state["current"]) # 设初始节点的提示与工具
else:
set_visible_tools(end_call_schemas)
set_visible_tools([])
await brain.setup(
cfg,
BrainRuntime(
context=context,
llm=llm,
queue_frame=worker.queue_frame,
set_system_prompt=set_system_prompt,
set_tools=set_visible_tools,
call_end=call_end,
),
)
async def append_user_text_to_context(text: str, *, run_llm: bool) -> None:
await worker.queue_frame(
@@ -862,19 +637,12 @@ async def run_pipeline(
@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})
brain.record_user_message(message.content)
await queue_transcript("user", message.content, message.timestamp)
@assistant_aggregator.event_handler("on_assistant_text_start")
async def on_assistant_text_start(_aggregator, turn_id, timestamp):
# 进入结束节点后,第一条「开始生成」的回复就是结束节点自己的结束语
# (其 text_start 发生在 ended 置位之后,不会误认转移前的那句)。
if (
workflow_active
and wf_state["ended"]
and wf_state["end_turn_id"] is None
):
wf_state["end_turn_id"] = turn_id
await brain.on_assistant_text_start(turn_id)
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message={
@@ -909,24 +677,12 @@ async def run_pipeline(
}
)
)
# 助手把话说完(未被打断)后:累加本节点轮次,必要时走文本兜底路由。
# 正常情况下转移由 LLM 直接调用转移工具完成(go_to_node),无需这里处理。
if content and not interrupted and workflow_active:
history.append({"role": "assistant", "content": content})
if turn_id == wf_state["end_turn_id"]:
# 结束节点的结束语文本已生成完(也已下发 data channel),武装挂断;
# 真正的 EndFrame 由 EndCallAfterSpeech 在结束语「说完」时排入。
wf_state["end_armed"] = True
elif not wf_state["ended"]:
wf_state["turns_in_node"] += 1
await fallback_route()
elif content and not interrupted:
history.append({"role": "assistant", "content": content})
await brain.on_assistant_text_end(turn_id, content, interrupted)
@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})
brain.record_user_message(text)
# 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。
await queue_transcript("user", text, time_now_iso8601())
@@ -941,7 +697,7 @@ async def run_pipeline(
@text_input.event_handler("on_text_append")
async def on_text_append(_processor, text):
# 静默追加:写进上下文但不打断、不触发推理;transcript 照常上报
history.append({"role": "user", "content": text})
brain.record_user_message(text)
await queue_transcript("user", text, time_now_iso8601())
await append_user_text_to_context(text, run_llm=False)
@@ -969,9 +725,7 @@ async def run_pipeline(
if brain.spec.owns_context:
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"])
await brain.on_connected()
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(_transport, _client):
@@ -996,12 +750,17 @@ async def run_realtime_pipeline(
transport,
cfg: AssistantConfig,
*,
brain: Brain,
assistant_id: str | None = None,
channel: str = "webrtc",
) -> None:
"""Run a speech-to-speech model that owns ASR, reasoning, and synthesis."""
realtime = create_realtime_service(cfg)
realtime = create_realtime_service(
cfg,
instructions=brain.system_prompt(cfg),
)
text_input = RealtimeTextInputProcessor()
greeting = await brain.greeting(cfg)
recorder = await ConversationRecorder.start(
assistant_id=assistant_id,
@@ -1058,8 +817,8 @@ async def run_realtime_pipeline(
@transport.event_handler("on_client_connected")
async def on_client_connected(_transport, _client):
if cfg.greeting:
await realtime.speak(cfg.greeting)
if greeting:
await realtime.speak(greeting)
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(_transport, _client):

4
backend/services/pipecat/service_factory.py
View File

@@ -142,7 +142,7 @@ def create_tts(cfg: AssistantConfig):
)
def create_realtime_service(cfg: AssistantConfig):
def create_realtime_service(cfg: AssistantConfig, *, instructions: str):
"""Create a speech-to-speech service that owns STT, LLM, and TTS."""
if cfg.realtime_interface_type == "stepfun-realtime":
from services.pipecat.stepfun_realtime import StepFunRealtimeService
@@ -151,7 +151,7 @@ def create_realtime_service(cfg: AssistantConfig):
api_key=_require(cfg.realtime_api_key, "Realtime apiKey"),
model=_require(cfg.realtimeModel, "Realtime modelId"),
base_url=_require(cfg.realtime_base_url, "Realtime apiUrl"),
instructions=cfg.prompt,
instructions=instructions,
voice=str(cfg.realtime_values.get("voice") or "linjiajiejie"),
input_sample_rate=int(
cfg.realtime_values.get("inputSampleRate") or 24000