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

Enhance workflow routing and agent configuration management

Browse Source 
- Introduce WorkflowLLMRouter for pre-response LLM routing, allowing agents to determine the appropriate function to call based on user input.
- Implement UserTurnRoutingProcessor to manage user turns before reaching the LLM, ensuring proper routing and handling of user messages.
- Refactor WorkflowBrain to integrate new routing logic and enhance agent stage configuration, including entry modes and resource management.
- Update service factory to support dynamic LLM resource configuration based on workflow settings.
- Add tests for new routing functionality and ensure proper handling of user messages in various scenarios.
This commit is contained in:
Xin Wang
2026-07-14 09:36:28 +08:00
parent 32aef14ddb
commit 72856bf3a7
19 changed files with 2611 additions and 552 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
backend/services/brains/workflow_brain.py
View File

@@ -15,6 +15,7 @@ from pipecat.flows import (
NodeConfig,
)
from pipecat.frames.frames import (
LLMRunFrame,
LLMUpdateSettingsFrame,
OutputTransportMessageUrgentFrame,
TTSSpeakFrame,
@@ -22,18 +23,20 @@ from pipecat.frames.frames import (
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.services.settings import LLMSettings
from pipecat.utils.time import time_now_iso8601
from services.brains.base import BaseBrain, BrainRuntime, BrainSpec
from services.knowledge import search as search_knowledge
from services.runtime_variables import DynamicVariableStore
from services.tool_executor import ToolExecutionError, ToolExecutor
from services.workflow_engine import WorkflowEngine
from services.workflow_router import STAY_ON_CURRENT_AGENT, WorkflowLLMRouter
MAX_AUTOMATIC_HOPS = 50
AGENT_STAGE_INSTRUCTION = (
"完成当前阶段任务。需要流转时必须调用对应的转移函数；"
"不要在调用转移函数后继续生成口头回复。"
"工作流路由已在用户一轮输入结束时完成。只执行当前阶段任务，"
"不要自行解释、模拟或宣布节点切换。"
)
@@ -58,6 +61,7 @@ class WorkflowBrain(BaseBrain):
}
self._runtime: BrainRuntime | None = None
self._manager: FlowManager | None = None
self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow"))
self._ended = False
async def greeting(self, cfg: AssistantConfig) -> str:
@@ -79,6 +83,7 @@ class WorkflowBrain(BaseBrain):
self._store = DynamicVariableStore.from_config(cfg)
self._tools = ToolExecutor(self._store)
self._tool_by_id = {tool.id: tool for tool in cfg.tools}
self._router = WorkflowLLMRouter(cfg)
self._manager = FlowManager(
worker=runtime.worker,
llm=runtime.llm,
@@ -95,9 +100,13 @@ class WorkflowBrain(BaseBrain):
self._store,
include_default=True,
)
if not edge:
if not edge and self._engine.has_outgoing(self._engine.start_id):
raise RuntimeError("Start 初始化后没有命中的表达式边或默认边")
node_config = await self._follow_edge(edge)
node_config = (
await self._follow_edge(edge)
if edge
else self._passive_node_config(self._engine.start_id)
)
if self._manager is None:
raise RuntimeError("Workflow FlowManager 尚未初始化")
await self._manager.initialize(node_config)
@@ -107,6 +116,76 @@ class WorkflowBrain(BaseBrain):
if content and not self._ended:
self._store.record("user", content)
async def on_user_turn_end(self, content: str) -> bool:
"""Route a complete user turn before any Agent is allowed to reply."""
if not content or self._ended:
return True
self.record_user_message(content)
manager = self._require_manager()
current = manager.current_node
if not current or self._engine.node_type(current) != "agent":
return True
edge = self._engine.deterministic_edge(
current,
self._store,
include_default=False,
)
outgoing = self._engine.outgoing(current)
llm_edges = [
candidate
for candidate in outgoing
if self._engine.edge_mode(candidate) == "llm"
]
default_edge = next(
(
candidate
for candidate in outgoing
if self._engine.edge_mode(candidate) == "always"
),
None,
)
if edge is None and llm_edges:
selected = await self._router_for_node(current).select_edge(
node_name=self._engine.name(current),
node_prompt=self._engine.prompt_for(current, self._store),
edges=llm_edges,
history=self._store.history,
variables={
key: value
for key, value in self._store.values.items()
if not key.startswith("system__")
},
edge_name=self._engine.edge_fn_name,
edge_description=self._engine.edge_description,
)
if selected and selected != STAY_ON_CURRENT_AGENT:
edge = next(
(
candidate
for candidate in llm_edges
if self._engine.edge_fn_name(candidate) == selected
),
None,
)
elif selected == STAY_ON_CURRENT_AGENT:
edge = default_edge
elif edge is None and not llm_edges:
edge = default_edge
if edge and manager.current_node == current:
next_config = await self._follow_edge(edge)
await manager.set_node_from_config(next_config)
return True
# The incoming LLMContextFrame is intentionally suppressed by the
# pipeline router. Queue prompt refresh + inference in this order so
# this user turn is answered with the current Agent's latest variables.
await self._refresh_agent_prompt(current)
await self._require_runtime().queue_frame(LLMRunFrame())
return True
async def on_assistant_text_end(
self,
_turn_id: str,
@@ -116,19 +195,6 @@ class WorkflowBrain(BaseBrain):
if not content or interrupted or self._ended:
return
self._store.record("agent", content, completed_agent_turn=True)
manager = self._require_manager()
current = manager.current_node
if not current or self._engine.node_type(current) != "agent":
return
await self._refresh_agent_prompt(current)
edge = self._engine.deterministic_edge(
current,
self._store,
include_default=False,
)
if edge and manager.current_node == current:
next_config = await self._follow_edge(edge)
await manager.set_node_from_config(next_config)
async def _refresh_agent_prompt(self, node_id: str) -> None:
runtime = self._require_runtime()
@@ -145,61 +211,104 @@ class WorkflowBrain(BaseBrain):
stage_prompt = self._engine.prompt_for(node_id, self._store)
return f"{stage_prompt}\n\n[工作流执行规则]\n{AGENT_STAGE_INSTRUCTION}"
def _router_for_node(self, node_id: str) -> WorkflowLLMRouter:
stage = self._engine.agent_stage_config(node_id)
resource_id = stage.llm_resource_id
cfg = self._cfg
resource = cfg.workflow_model_resources.get(resource_id) if cfg else None
if not cfg or not resource:
return self._router
from services.pipecat.service_factory import config_with_resource
return WorkflowLLMRouter(config_with_resource(cfg, resource))
async def _apply_agent_stage(self, node_id: str) -> None:
data = self._engine.data(node_id)
stage = self._engine.agent_stage_config(node_id)
await self._emit_node_active(node_id)
if self._runtime and self._runtime.set_input_enabled:
self._runtime.set_input_enabled(True)
asr_id = str(
data.get("asrResourceId")
or self._engine.settings.get("defaultAsrResourceId")
or ""
)
tts_id = str(
data.get("ttsResourceId")
or self._engine.settings.get("defaultTtsResourceId")
or ""
)
runtime = self._require_runtime()
if runtime.switch_services:
await runtime.switch_services(asr_id or None, tts_id or None)
await runtime.switch_services(
stage.llm_resource_id or None,
stage.asr_resource_id or None,
stage.tts_resource_id or None,
)
if runtime.set_knowledge_scope:
runtime.set_knowledge_scope(
{
"knowledge_base_id": data.get("knowledgeBaseId"),
"mode": data.get("knowledgeMode", "disabled"),
"top_n": int(data.get("knowledgeTopN") or 5),
"score_threshold": float(data.get("knowledgeScoreThreshold") or 0.0),
"knowledge_base_id": stage.knowledge_base_id,
"mode": stage.knowledge_mode,
"top_n": stage.knowledge_top_n,
"score_threshold": stage.knowledge_score_threshold,
}
)
def _agent_config(self, node_id: str) -> NodeConfig:
data = self._engine.data(node_id)
entry_mode = str(data.get("entryMode") or "wait_user")
entry_speech = self._store.render(str(data.get("entrySpeech") or ""))
strategy = (
ContextStrategy.RESET
if data.get("contextPolicy") == "fresh"
else ContextStrategy.APPEND
)
stage = self._engine.agent_stage_config(node_id)
functions: list[FlowsFunctionSchema] = []
for tool_id in data.get("toolIds") or []:
for tool_id in stage.tool_ids:
tool = self._tool_by_id.get(str(tool_id))
if tool and tool.type == "http":
functions.append(self._flow_tool(tool, node_id))
knowledge_function = self._knowledge_function(node_id)
if knowledge_function:
functions.append(knowledge_function)
for edge in self._engine.llm_edges(node_id):
functions.append(self._flow_edge(edge))
return {
config: NodeConfig = {
"name": node_id,
"role_message": self._agent_role_message(node_id),
"task_messages": [],
"task_messages": (
[{"role": "assistant", "content": entry_speech}]
if entry_mode == "fixed_speech"
else []
),
"functions": functions,
"context_strategy": ContextStrategyConfig(strategy=strategy),
"respond_immediately": True,
"respond_immediately": entry_mode == "generate",
}
if entry_mode == "fixed_speech":
config["pre_actions"] = [
{
"type": "workflow_fixed_speech",
"text": entry_speech,
"handler": self._play_fixed_speech,
}
]
return config
def _terminal_config(self, node_id: str) -> NodeConfig:
async def _play_fixed_speech(self, action: dict, _flow_manager: FlowManager) -> None:
"""Play and persist Agent entry speech without creating an LLM turn."""
await self._queue_visible_speech(str(action.get("text") or ""))
async def _queue_visible_speech(self, text: str) -> None:
"""Show and persist fixed workflow speech before sending it to TTS."""
content = text.strip()
if not content:
return
self._store.record("agent", content)
runtime = self._require_runtime()
await runtime.queue_frame(
OutputTransportMessageUrgentFrame(
message={
"type": "transcript",
"role": "assistant",
"content": content,
"timestamp": time_now_iso8601(),
}
)
)
await runtime.queue_frame(TTSSpeakFrame(content, append_to_context=False))
def _passive_node_config(self, node_id: str) -> NodeConfig:
"""Keep a non-conversational terminal node active without ending the call."""
return {
"name": node_id,
"role_message": self._store.render(self._engine.global_prompt()),
@@ -235,24 +344,13 @@ class WorkflowBrain(BaseBrain):
properties=properties,
required=required,
handler=handler,
)
def _flow_edge(self, edge: dict) -> FlowsFunctionSchema:
async def handler(_args, _flow_manager):
return None, await self._follow_edge(edge)
return FlowsFunctionSchema(
name=self._engine.edge_fn_name(edge),
description=self._engine.edge_description(edge),
properties={},
required=[],
handler=handler,
cancel_on_interruption=True,
)
def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None:
data = self._engine.data(node_id)
knowledge_id = str(data.get("knowledgeBaseId") or "")
if not knowledge_id or data.get("knowledgeMode") != "on_demand":
stage = self._engine.agent_stage_config(node_id)
knowledge_id = str(stage.knowledge_base_id or "")
if not knowledge_id or stage.knowledge_mode != "on_demand":
return None
cfg = self._cfg or AssistantConfig(type="workflow")
knowledge = cfg.workflow_knowledge_bases.get(knowledge_id)
@@ -270,8 +368,8 @@ class WorkflowBrain(BaseBrain):
session,
knowledge_id,
query,
top_k=int(data.get("knowledgeTopN") or 5),
score_threshold=float(data.get("knowledgeScoreThreshold") or 0.0),
top_k=stage.knowledge_top_n,
score_threshold=stage.knowledge_score_threshold,
)
return {"status": "ok", "results": results}
except Exception as exc: # noqa: BLE001 - tool errors are returned to the LLM
@@ -286,14 +384,13 @@ class WorkflowBrain(BaseBrain):
},
required=["query"],
handler=handler,
cancel_on_interruption=True,
)
async def _follow_edge(self, edge: dict) -> NodeConfig:
speech = self._engine.edge_transition_speech(edge)
if speech:
await self._require_runtime().queue_frame(
TTSSpeakFrame(self._store.render(speech), append_to_context=False)
)
await self._queue_visible_speech(self._store.render(speech))
return await self._resolve_path(str(edge.get("target") or ""))
async def _resolve_path(self, node_id: str) -> NodeConfig:
@@ -304,7 +401,7 @@ class WorkflowBrain(BaseBrain):
return self._agent_config(node_id)
if node_type == "end":
await self._enter_end(node_id)
return self._terminal_config(node_id)
return self._passive_node_config(node_id)
if node_type == "action":
await self._enter_action(node_id)
elif node_type == "handoff":
@@ -313,6 +410,8 @@ class WorkflowBrain(BaseBrain):
await self._emit_node_active(node_id)
else:
raise RuntimeError(f"工作流指向未知节点:{node_id}")
if not self._engine.has_outgoing(node_id):
return self._passive_node_config(node_id)
edge = self._engine.deterministic_edge(
node_id,
self._store,
@@ -322,9 +421,7 @@ class WorkflowBrain(BaseBrain):
raise RuntimeError(f"自动节点 {node_id} 没有命中的表达式边或默认边")
speech = self._engine.edge_transition_speech(edge)
if speech:
await self._require_runtime().queue_frame(
TTSSpeakFrame(self._store.render(speech), append_to_context=False)
)
await self._queue_visible_speech(self._store.render(speech))
node_id = str(edge.get("target") or "")
raise RuntimeError("工作流连续自动跳转超过安全上限")
@@ -366,9 +463,7 @@ class WorkflowBrain(BaseBrain):
)
)
if message:
await self._require_runtime().queue_frame(
TTSSpeakFrame(message, append_to_context=False)
)
await self._queue_visible_speech(message)
self._store.values["system__handoff_status"] = "requested"
async def _enter_end(self, node_id: str) -> None:
@@ -389,12 +484,12 @@ class WorkflowBrain(BaseBrain):
)
)
if message:
await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False))
await self._queue_visible_speech(message)
return
runtime.call_end.begin("workflow_completed")
if message:
runtime.call_end.arm_after_speech()
await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False))
await self._queue_visible_speech(message)
else:
await runtime.call_end.finish()