1306 lines
61 KiB
Python
1306 lines
61 KiB
Python
from __future__ import annotations
|
||
|
||
import re
|
||
from time import perf_counter
|
||
from uuid import uuid4
|
||
|
||
from app.plugins.base import PluginRegistry
|
||
from app.schemas.chat import ChatRequest, ChatResponse, FillSlotsRequest
|
||
from app.schemas.debug import IntentCandidate, MatcherStageDebug, RoutingDebug
|
||
from app.schemas.workflow import MissingSlot, Workflow, WorkflowStep
|
||
from app.services.dialog_act import DialogActEngine
|
||
from app.services.dialog_rules import DialogRuleEngine
|
||
from app.services.intent_registry import IntentRegistry
|
||
from app.services.knowledge_llm import DashScopeKnowledgeLLM
|
||
from app.services.planner import PlanningResult, WorkflowPlanner
|
||
from app.services.response_policy import ResponsePolicy
|
||
from app.services.rewrite_engine import ContextRewriteEngine, RewriteResult
|
||
from app.services.router import Router
|
||
from app.services.session_store import InMemorySessionStore, SessionState
|
||
from app.services.social import SocialResponder, SocialRouter
|
||
|
||
|
||
class AgentService:
|
||
_MUSIC_GENRE_HINTS = ("轻音乐", "摇滚", "古典", "民谣", "爵士", "流行", "儿歌")
|
||
|
||
def __init__(
|
||
self,
|
||
intent_registry: IntentRegistry,
|
||
router: Router,
|
||
plugins: PluginRegistry,
|
||
session_store: InMemorySessionStore | None = None,
|
||
rewrite_engine: ContextRewriteEngine | None = None,
|
||
response_policy: ResponsePolicy | None = None,
|
||
dialog_rules: DialogRuleEngine | None = None,
|
||
dialog_act_engine: DialogActEngine | None = None,
|
||
planner: WorkflowPlanner | None = None,
|
||
social_router: SocialRouter | None = None,
|
||
social_responder: SocialResponder | None = None,
|
||
knowledge_llm: DashScopeKnowledgeLLM | None = None,
|
||
) -> None:
|
||
self.session_store = session_store or InMemorySessionStore()
|
||
self.intent_registry = intent_registry
|
||
self.router = router
|
||
self.plugins = plugins
|
||
self.rewrite_engine = rewrite_engine or ContextRewriteEngine()
|
||
self.response_policy = response_policy or ResponsePolicy()
|
||
self.dialog_rules = dialog_rules or DialogRuleEngine()
|
||
self.dialog_act_engine = dialog_act_engine or DialogActEngine()
|
||
self.planner = planner
|
||
self.social_router = social_router or SocialRouter()
|
||
self.social_responder = social_responder
|
||
self.knowledge_llm = knowledge_llm
|
||
|
||
def handle_chat(self, request: ChatRequest) -> ChatResponse:
|
||
started_at = perf_counter()
|
||
breakdown: dict[str, float] = {}
|
||
|
||
timer_started_at = perf_counter()
|
||
session = self.session_store.get_or_create(
|
||
session_id=request.session_id,
|
||
user_id=request.user_id,
|
||
channel=request.channel,
|
||
)
|
||
self._mark_timing(breakdown, "session_get_or_create_ms", timer_started_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
self._update_dialog_act(session, request.input_text)
|
||
self._mark_timing(breakdown, "dialog_act_ms", timer_started_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
stop_response = self._handle_stop_request(session, request.input_text)
|
||
self._mark_timing(breakdown, "stop_check_ms", timer_started_at)
|
||
if stop_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, stop_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(stop_response, started_at, breakdown, first_response_at)
|
||
|
||
if session.status == "waiting_confirmation":
|
||
confirmation = self._parse_confirmation_decision(request.input_text)
|
||
if confirmation is not None:
|
||
timer_started_at = perf_counter()
|
||
response = self._handle_confirmation(session, request.input_text)
|
||
self._mark_timing(breakdown, "confirmation_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
continued_response = self._continue_active_task(session, request.input_text)
|
||
self._mark_timing(breakdown, "active_task_ms", timer_started_at)
|
||
if continued_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, continued_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(continued_response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
social_response = self._handle_social_turn(session, request.input_text)
|
||
self._mark_timing(breakdown, "social_route_ms", timer_started_at)
|
||
if social_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, social_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(social_response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
rewrite_result = self.rewrite_engine.rewrite(request.input_text, session)
|
||
rewrite_elapsed_ms = round((perf_counter() - timer_started_at) * 1000, 3)
|
||
breakdown["rewrite_ms"] = rewrite_elapsed_ms
|
||
|
||
timer_started_at = perf_counter()
|
||
route_result = self.router.route(rewrite_result.rewritten_text)
|
||
self._mark_timing(breakdown, "route_ms", timer_started_at)
|
||
self._attach_rewrite_debug(route_result.debug, rewrite_result, rewrite_elapsed_ms)
|
||
|
||
timer_started_at = perf_counter()
|
||
# ── BERT 信号不稳时，知识库优先检索（在 Planner 之前）──────────────────
|
||
# decision=route_to_cloud 表示 BERT 低置信，需发给 cloud planner
|
||
# 但如果本地 MD 知识库关键词已命中，直接走 knowledge_llm，跳过 cloud planner
|
||
if route_result.debug.decision == "route_to_cloud" and self.knowledge_llm is not None:
|
||
quick_hits = self.knowledge_llm._store.search(rewrite_result.rewritten_text, top_k=1)
|
||
if quick_hits and quick_hits[0].score >= 3.0: # 至少命中一个 title 或多个 body 词
|
||
early_knowledge = self._try_knowledge_llm(
|
||
session, request.input_text, route_result.debug, rewrite_result
|
||
)
|
||
if early_knowledge is not None:
|
||
self._mark_timing(breakdown, "knowledge_llm_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
self._record_turn(session, request.input_text, early_knowledge.reply_text, rewrite_result)
|
||
self.session_store.save(session)
|
||
return self._finalize_response(early_knowledge, started_at, breakdown, first_response_at)
|
||
planning_result = self._plan_if_needed(session, rewrite_result.rewritten_text, route_result.debug)
|
||
self._mark_timing(breakdown, "planner_ms", timer_started_at)
|
||
if planning_result is not None and planning_result.accepted and planning_result.steps:
|
||
timer_started_at = perf_counter()
|
||
response = self._start_planned_workflow(session, planning_result, route_result.debug, request.input_text, rewrite_result)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
decision_response = self._handle_route_decision(session, route_result.debug, request.input_text, rewrite_result)
|
||
self._mark_timing(breakdown, "decision_response_ms", timer_started_at)
|
||
if decision_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(decision_response, started_at, breakdown, first_response_at)
|
||
intent = route_result.intent
|
||
if intent is None:
|
||
timer_started_at = perf_counter()
|
||
# ── BERT 未命中：尝试 LLM + knowledge_search function call 兆底 ──
|
||
knowledge_response = self._try_knowledge_llm(
|
||
session, request.input_text, route_result.debug, rewrite_result
|
||
)
|
||
if knowledge_response is not None:
|
||
self._mark_timing(breakdown, "knowledge_llm_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
self._record_turn(session, request.input_text, knowledge_response.reply_text, rewrite_result)
|
||
self.session_store.save(session)
|
||
return self._finalize_response(knowledge_response, started_at, breakdown, first_response_at)
|
||
response = self._fallback_response(session.session_id, route_result.debug)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, rewrite_result)
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
if session.current_intent != intent.intent_id:
|
||
session.pending_slots = []
|
||
session.slots = {}
|
||
session.workflow = None
|
||
|
||
session.current_intent = intent.intent_id
|
||
session.status = "understanding"
|
||
timer_started_at = perf_counter()
|
||
extracted_slots = self.router.extract_slots(rewrite_result.rewritten_text, intent)
|
||
self._mark_timing(breakdown, "slot_extract_ms", timer_started_at)
|
||
route_result.debug.extracted_slots = extracted_slots.copy()
|
||
session.routing_debug = route_result.debug.model_dump()
|
||
session.slots.update(extracted_slots)
|
||
self._update_context_memory(session, intent.intent_id, extracted_slots)
|
||
|
||
timer_started_at = perf_counter()
|
||
response = self._build_response_from_session(session)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, rewrite_result)
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
def _handle_route_decision(
|
||
self,
|
||
session: SessionState,
|
||
routing_debug: RoutingDebug,
|
||
user_text: str,
|
||
rewrite_result: RewriteResult,
|
||
) -> ChatResponse | None:
|
||
if routing_debug.decision == "execute":
|
||
return None
|
||
if routing_debug.decision == "clarify":
|
||
response = ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="clarify",
|
||
reply_text=self.response_policy.clarify(self._top_candidate_ids(routing_debug)),
|
||
intent=routing_debug.selected_intent,
|
||
domain=self._domain_for_intent(routing_debug.selected_intent),
|
||
decision=routing_debug.decision,
|
||
decision_reason=routing_debug.decision_reason,
|
||
status="clarify",
|
||
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
self._record_turn(session, user_text, response.reply_text, rewrite_result)
|
||
return response
|
||
if routing_debug.decision == "reject":
|
||
response = ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="reject",
|
||
reply_text=self.response_policy.reject(),
|
||
decision=routing_debug.decision,
|
||
decision_reason=routing_debug.decision_reason,
|
||
status="rejected",
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
self._record_turn(session, user_text, response.reply_text, rewrite_result)
|
||
return response
|
||
if routing_debug.decision == "route_to_cloud":
|
||
candidate_ids = self._top_candidate_ids(routing_debug)
|
||
planner_stage = next((stage for stage in reversed(routing_debug.stages) if stage.stage == "planner"), None)
|
||
planner_reason = planner_stage.reason if planner_stage is not None else routing_debug.decision_reason
|
||
if self._planner_indicates_out_of_scope(planner_stage):
|
||
response = ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="reject",
|
||
reply_text=self.response_policy.reject(),
|
||
intent=routing_debug.selected_intent,
|
||
domain=self._domain_for_intent(routing_debug.selected_intent),
|
||
decision="reject",
|
||
decision_reason=planner_reason,
|
||
|
||
status="rejected",
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
elif candidate_ids:
|
||
response = ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="clarify",
|
||
reply_text=self.response_policy.clarify(candidate_ids),
|
||
intent=routing_debug.selected_intent,
|
||
domain=self._domain_for_intent(routing_debug.selected_intent),
|
||
decision=routing_debug.decision,
|
||
decision_reason=planner_reason,
|
||
|
||
status="route_to_cloud",
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
else:
|
||
response = ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="fallback",
|
||
reply_text=self.response_policy.fallback(),
|
||
intent=routing_debug.selected_intent,
|
||
domain=self._domain_for_intent(routing_debug.selected_intent),
|
||
decision=routing_debug.decision,
|
||
decision_reason=planner_reason,
|
||
|
||
status="route_to_cloud",
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
self._record_turn(session, user_text, response.reply_text, rewrite_result)
|
||
return response
|
||
return None
|
||
|
||
def handle_fill_slots(self, request: FillSlotsRequest) -> ChatResponse:
|
||
started_at = perf_counter()
|
||
breakdown: dict[str, float] = {}
|
||
|
||
timer_started_at = perf_counter()
|
||
session = self.session_store.get(request.session_id)
|
||
self._mark_timing(breakdown, "session_get_ms", timer_started_at)
|
||
if session is None or session.current_intent is None:
|
||
timer_started_at = perf_counter()
|
||
response = ChatResponse(
|
||
session_id=request.session_id,
|
||
reply_type="fallback",
|
||
reply_text=self.response_policy.fallback(),
|
||
status="fallback",
|
||
routing_debug=None,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
if session is not None:
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
self._update_dialog_act(session, request.input_text)
|
||
self._mark_timing(breakdown, "dialog_act_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
stop_response = self._handle_stop_request(session, request.input_text)
|
||
self._mark_timing(breakdown, "stop_check_ms", timer_started_at)
|
||
if stop_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, stop_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(stop_response, started_at, breakdown, first_response_at)
|
||
|
||
if session.status == "waiting_confirmation":
|
||
if self._parse_confirmation_decision(request.input_text) is not None:
|
||
timer_started_at = perf_counter()
|
||
response = self._handle_confirmation(session, request.input_text)
|
||
self._mark_timing(breakdown, "confirmation_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
continued_response = self._continue_active_task(session, request.input_text)
|
||
self._mark_timing(breakdown, "active_task_ms", timer_started_at)
|
||
if continued_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, continued_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(continued_response, started_at, breakdown, first_response_at)
|
||
|
||
timer_started_at = perf_counter()
|
||
social_response = self._handle_social_turn(session, request.input_text)
|
||
self._mark_timing(breakdown, "social_route_ms", timer_started_at)
|
||
if social_response is not None:
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, social_response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(social_response, started_at, breakdown, first_response_at)
|
||
|
||
if session.status == "waiting_confirmation":
|
||
timer_started_at = perf_counter()
|
||
response = self._handle_confirmation(session, request.input_text)
|
||
self._mark_timing(breakdown, "confirmation_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
existing_workflow = self._workflow_from_session(session)
|
||
if existing_workflow is not None and existing_workflow.workflow_type != "single":
|
||
intent = self.intent_registry.get(session.current_intent)
|
||
timer_started_at = perf_counter()
|
||
extracted_slots = self._extract_continuation_slots(session, intent, request.input_text)
|
||
self._mark_timing(breakdown, "slot_extract_ms", timer_started_at)
|
||
session.slots.update(extracted_slots)
|
||
self._update_context_memory(session, intent.intent_id, extracted_slots)
|
||
timer_started_at = perf_counter()
|
||
response = self._continue_planned_workflow(
|
||
session,
|
||
existing_workflow,
|
||
self._routing_debug_from_session(session),
|
||
)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
intent = self.intent_registry.get(session.current_intent)
|
||
timer_started_at = perf_counter()
|
||
extracted_slots = self._extract_continuation_slots(session, intent, request.input_text)
|
||
self._mark_timing(breakdown, "slot_extract_ms", timer_started_at)
|
||
session.slots.update(extracted_slots)
|
||
self._update_context_memory(session, intent.intent_id, extracted_slots)
|
||
if session.routing_debug is not None:
|
||
routing_debug = RoutingDebug.model_validate(session.routing_debug)
|
||
routing_debug.extracted_slots = session.slots.copy()
|
||
session.routing_debug = routing_debug.model_dump()
|
||
timer_started_at = perf_counter()
|
||
response = self._build_response_from_session(session)
|
||
self._mark_timing(breakdown, "response_build_ms", timer_started_at)
|
||
first_response_at = perf_counter()
|
||
timer_started_at = perf_counter()
|
||
self._record_turn(session, request.input_text, response.reply_text, RewriteResult(request.input_text, request.input_text))
|
||
self._mark_timing(breakdown, "record_turn_ms", timer_started_at)
|
||
timer_started_at = perf_counter()
|
||
self.session_store.save(session)
|
||
self._mark_timing(breakdown, "session_save_ms", timer_started_at)
|
||
return self._finalize_response(response, started_at, breakdown, first_response_at)
|
||
|
||
def _build_response_from_session(self, session: SessionState) -> ChatResponse:
|
||
assert session.current_intent is not None
|
||
intent = self.intent_registry.get(session.current_intent)
|
||
session.slots = self._normalize_slots_for_intent(intent.intent_id, session.slots)
|
||
pending_slots = self._pending_slots_for_intent(intent, session.slots)
|
||
|
||
if pending_slots:
|
||
session.pending_slots = pending_slots
|
||
session.status = "waiting_slot"
|
||
self.session_store.save(session)
|
||
first_slot = pending_slots[0]
|
||
workflow = self._build_workflow(session, pending_slots)
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="ask_slot",
|
||
reply_text=self.response_policy.ask_for_slot(
|
||
intent,
|
||
first_slot,
|
||
self._default_ask_template(intent, first_slot),
|
||
),
|
||
intent=session.current_intent,
|
||
domain=self._domain_for_intent(session.current_intent),
|
||
status=session.status,
|
||
pending_slots=pending_slots,
|
||
filled_slots=session.slots,
|
||
|
||
workflow=workflow,
|
||
routing_debug=self._routing_debug_from_session(session),
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
plugin_result = self.plugins.execute(intent.plugin_id, session.slots)
|
||
session.pending_slots = []
|
||
session.status = "completed"
|
||
workflow = self._build_workflow(session, [])
|
||
workflow.status = "completed"
|
||
self.session_store.save(session)
|
||
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="workflow_result",
|
||
reply_text=self.response_policy.workflow_result(intent, plugin_result),
|
||
intent=session.current_intent,
|
||
domain=self._domain_for_intent(session.current_intent),
|
||
status=session.status,
|
||
pending_slots=[],
|
||
filled_slots=session.slots,
|
||
|
||
workflow=workflow,
|
||
routing_debug=self._routing_debug_from_session(session),
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
def _build_workflow(self, session: SessionState, pending_slots: list[str]) -> Workflow:
|
||
assert session.current_intent is not None
|
||
intent = self.intent_registry.get(session.current_intent)
|
||
missing = [
|
||
MissingSlot(
|
||
slot_name=slot,
|
||
ask_template=self._default_ask_template(intent, slot),
|
||
priority=index + 1,
|
||
)
|
||
for index, slot in enumerate(pending_slots)
|
||
]
|
||
step = WorkflowStep(
|
||
step=1,
|
||
step_id=f"step_{session.current_intent}_1",
|
||
intent_id=session.current_intent,
|
||
plugin_id=intent.plugin_id,
|
||
action=intent.plugin_id.split(".")[-1],
|
||
status="pending" if pending_slots else "completed",
|
||
slots=session.slots.copy(),
|
||
)
|
||
workflow = Workflow(
|
||
workflow_id=f"wf_{session.session_id}",
|
||
workflow_type="single",
|
||
domain=intent.domain,
|
||
intent_id=session.current_intent,
|
||
status="waiting_slot" if pending_slots else "ready",
|
||
risk_level=intent.risk_level,
|
||
slots=session.slots.copy(),
|
||
missing_slots=missing,
|
||
steps=[step],
|
||
meta={
|
||
"source": "rule_router",
|
||
"routing_debug": session.routing_debug or {},
|
||
},
|
||
)
|
||
session.workflow = workflow.model_dump()
|
||
return workflow
|
||
|
||
def _fallback_response(self, session_id: str, routing_debug: RoutingDebug | None = None) -> ChatResponse:
|
||
return ChatResponse(
|
||
session_id=session_id,
|
||
reply_type="fallback",
|
||
reply_text=self.response_policy.fallback(),
|
||
decision=routing_debug.decision if routing_debug is not None else None,
|
||
decision_reason=routing_debug.decision_reason if routing_debug is not None else None,
|
||
status="fallback",
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
def _handle_stop_request(self, session: SessionState, text: str) -> ChatResponse | None:
|
||
if not self._is_stop_request(text) or not self._has_active_task(session):
|
||
return None
|
||
self._reset_active_task(session)
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="text",
|
||
reply_text=self.response_policy.task_stopped(),
|
||
status="stopped",
|
||
pending_slots=[],
|
||
filled_slots={},
|
||
workflow=None,
|
||
routing_debug=None,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
def _handle_social_turn(self, session: SessionState, text: str) -> ChatResponse | None:
|
||
social_route = self.social_router.route(text, session)
|
||
if social_route.category == "none":
|
||
return None
|
||
if self.social_responder is None:
|
||
reply_text = self.response_policy.open_social_fallback()
|
||
return self._social_response(
|
||
session=session,
|
||
reply_text=reply_text,
|
||
decision="open_social",
|
||
decision_reason="open social matched but no responder configured",
|
||
)
|
||
social_result = self.social_responder.reply(text, session)
|
||
reply_text = social_result.text.strip() or self.response_policy.open_social_fallback()
|
||
return self._social_response(
|
||
session=session,
|
||
reply_text=reply_text,
|
||
decision="open_social",
|
||
decision_reason=social_route.reason,
|
||
)
|
||
|
||
def _social_response(
|
||
self,
|
||
session: SessionState,
|
||
reply_text: str,
|
||
decision: str,
|
||
decision_reason: str,
|
||
) -> ChatResponse:
|
||
session.context_memory["last_dialog_mode"] = decision
|
||
pending_hint = self.response_policy.pending_task_hint(
|
||
session.status,
|
||
session.pending_slots,
|
||
session.current_intent,
|
||
)
|
||
final_reply = self.response_policy.with_pending_hint(reply_text, pending_hint)
|
||
workflow = self._workflow_from_session(session)
|
||
status = session.status if session.status != "idle" else "social"
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="text",
|
||
reply_text=final_reply,
|
||
intent=session.current_intent,
|
||
domain=self._domain_for_intent(session.current_intent),
|
||
decision=decision,
|
||
decision_reason=decision_reason,
|
||
|
||
status=status,
|
||
pending_slots=list(session.pending_slots),
|
||
filled_slots=session.slots.copy(),
|
||
workflow=workflow,
|
||
routing_debug=self._routing_debug_from_session(session),
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
def _routing_debug_from_session(self, session: SessionState) -> RoutingDebug | None:
|
||
if session.routing_debug is None:
|
||
return None
|
||
return RoutingDebug.model_validate(session.routing_debug)
|
||
|
||
def _trace_id(self) -> str:
|
||
return uuid4().hex
|
||
|
||
def _domain_for_intent(self, intent_id: str | None) -> str | None:
|
||
"""Return the domain for a given intent_id, or None if unknown."""
|
||
if intent_id is None:
|
||
return None
|
||
try:
|
||
return self.intent_registry.get(intent_id).domain
|
||
except KeyError:
|
||
return None
|
||
|
||
def _planner_indicates_out_of_scope(self, planner_stage: MatcherStageDebug | None) -> bool:
|
||
if planner_stage is None or planner_stage.accepted:
|
||
return False
|
||
reason = str(planner_stage.reason or "").strip().lower()
|
||
if not reason:
|
||
return False
|
||
out_of_scope_signals = (
|
||
"no matching intent",
|
||
"there is no matching intent",
|
||
"not in the catalog",
|
||
"intent catalog only contains",
|
||
"provided intent catalog only contains",
|
||
"out of scope",
|
||
"third-party app action",
|
||
"outside known local capabilities",
|
||
"没有匹配意图",
|
||
"不在意图目录",
|
||
"不在目录中",
|
||
"超出能力范围",
|
||
"能力之外",
|
||
)
|
||
return any(signal in reason for signal in out_of_scope_signals)
|
||
|
||
def _mark_timing(self, breakdown: dict[str, float], name: str, started_at: float) -> None:
|
||
breakdown[name] = round((perf_counter() - started_at) * 1000, 3)
|
||
|
||
def _finalize_response(
|
||
self,
|
||
response: ChatResponse,
|
||
started_at: float,
|
||
breakdown: dict[str, float],
|
||
first_response_at: float | None = None,
|
||
) -> ChatResponse:
|
||
first_ready_at = first_response_at or perf_counter()
|
||
response.first_response_latency_ms = round((first_ready_at - started_at) * 1000, 3)
|
||
response.total_latency_ms = round((perf_counter() - started_at) * 1000, 3)
|
||
response.processing_breakdown = dict(breakdown)
|
||
if response.routing_debug is not None:
|
||
if response.routing_debug.total_match_latency_ms is not None:
|
||
response.processing_breakdown.setdefault(
|
||
"match_pipeline_ms",
|
||
response.routing_debug.total_match_latency_ms,
|
||
)
|
||
return response
|
||
|
||
def _top_candidate_ids(self, routing_debug: RoutingDebug, limit: int = 3) -> list[str]:
|
||
if not routing_debug.stages:
|
||
return []
|
||
fusion_stage = next((stage for stage in reversed(routing_debug.stages) if stage.stage == "fusion"), None)
|
||
if fusion_stage is None:
|
||
return []
|
||
return [candidate.intent_id for candidate in fusion_stage.candidates[:limit]]
|
||
|
||
def _plan_if_needed(
|
||
self,
|
||
session: SessionState,
|
||
text: str,
|
||
routing_debug: RoutingDebug,
|
||
) -> PlanningResult | None:
|
||
if self.planner is None:
|
||
return None
|
||
if not self._should_use_planner(text, routing_debug):
|
||
return None
|
||
planner_started_at = perf_counter()
|
||
result = self.planner.plan(
|
||
text,
|
||
self.intent_registry.list(),
|
||
context={
|
||
"current_intent": session.current_intent,
|
||
"slots": session.slots,
|
||
"context_memory": session.context_memory,
|
||
},
|
||
)
|
||
planner_elapsed_ms = round((perf_counter() - planner_started_at) * 1000, 3)
|
||
self._attach_planner_debug(routing_debug, result, planner_elapsed_ms)
|
||
return result
|
||
|
||
def _should_use_planner(self, text: str, routing_debug: RoutingDebug) -> bool:
|
||
fusion_stage = routing_debug.stages[-1] if routing_debug.stages else None
|
||
has_complex_pattern = any(token in text for token in ("然后", "并且", "同时", "如果", "若", "先", "后", "顺便", "以及"))
|
||
if has_complex_pattern:
|
||
return True
|
||
if self._looks_like_parallel_request(text, fusion_stage):
|
||
return True
|
||
if text.count("，") >= 1 or text.count(",") >= 1 or text.count("；") >= 1:
|
||
return True
|
||
if fusion_stage is not None and fusion_stage.stage == "fusion" and not fusion_stage.accepted:
|
||
return True
|
||
return False
|
||
|
||
def _attach_planner_debug(
|
||
self,
|
||
routing_debug: RoutingDebug,
|
||
planning_result: PlanningResult,
|
||
elapsed_ms: float | None = None,
|
||
) -> None:
|
||
metadata = {
|
||
"workflow_type": planning_result.workflow_type,
|
||
"raw_response": planning_result.raw_response,
|
||
}
|
||
metadata.update(planning_result.metadata)
|
||
routing_debug.stages.append(
|
||
MatcherStageDebug(
|
||
stage="planner",
|
||
accepted=planning_result.accepted,
|
||
selected_intent=planning_result.steps[0].intent_id if planning_result.steps else None,
|
||
score=float(len(planning_result.steps)),
|
||
elapsed_ms=elapsed_ms,
|
||
reason=planning_result.reason,
|
||
model_name=planning_result.model_name,
|
||
backend=planning_result.backend,
|
||
error_message=planning_result.error_message,
|
||
metadata=metadata,
|
||
candidates=[
|
||
IntentCandidate(
|
||
intent_id=step.intent_id,
|
||
score=1.0,
|
||
reason=step.reason,
|
||
model_name=planning_result.model_name,
|
||
metadata=planning_result.metadata.get("normalized_steps", [{}])[index]
|
||
if index < len(planning_result.metadata.get("normalized_steps", []))
|
||
else {"slots": step.slots},
|
||
)
|
||
for index, step in enumerate(planning_result.steps[:5])
|
||
],
|
||
)
|
||
)
|
||
|
||
def _start_planned_workflow(
|
||
self,
|
||
session: SessionState,
|
||
planning_result: PlanningResult,
|
||
routing_debug: RoutingDebug,
|
||
user_text: str,
|
||
rewrite_result: RewriteResult,
|
||
) -> ChatResponse:
|
||
workflow = self._build_planned_workflow(session, planning_result)
|
||
session.workflow = workflow.model_dump()
|
||
session.routing_debug = routing_debug.model_dump()
|
||
response = self._continue_planned_workflow(session, workflow, routing_debug)
|
||
self._record_turn(session, user_text, response.reply_text, rewrite_result)
|
||
return response
|
||
|
||
def _build_planned_workflow(self, session: SessionState, planning_result: PlanningResult) -> Workflow:
|
||
steps: list[WorkflowStep] = []
|
||
first_intent = planning_result.steps[0].intent_id
|
||
domain = self.intent_registry.get(first_intent).domain
|
||
risk_level = self.intent_registry.get(first_intent).risk_level
|
||
step_id_map = {index: f"step_{session.session_id}_{index}" for index in range(1, len(planning_result.steps) + 1)}
|
||
for index, planned_step in enumerate(planning_result.steps, start=1):
|
||
intent = self.intent_registry.get(planned_step.intent_id)
|
||
steps.append(
|
||
WorkflowStep(
|
||
step=index,
|
||
step_id=step_id_map[index],
|
||
intent_id=planned_step.intent_id,
|
||
plugin_id=intent.plugin_id,
|
||
action=intent.plugin_id.split(".")[-1],
|
||
status="pending",
|
||
depends_on=[
|
||
step_id_map[dep_index]
|
||
for dep_index in planned_step.depends_on
|
||
if dep_index in step_id_map
|
||
],
|
||
slots=dict(planned_step.slots),
|
||
condition=dict(planned_step.condition),
|
||
requires_confirmation=planned_step.requires_confirmation or self._requires_confirmation(intent),
|
||
)
|
||
)
|
||
return Workflow(
|
||
workflow_id=f"wf_{session.session_id}",
|
||
workflow_type=planning_result.workflow_type if planning_result.workflow_type in {"single", "sequence", "conditional", "parallel"} else "sequence",
|
||
domain=domain,
|
||
intent_id=first_intent,
|
||
status="ready",
|
||
risk_level=risk_level,
|
||
slots=session.slots.copy(),
|
||
missing_slots=[],
|
||
steps=steps,
|
||
meta={
|
||
"source": planning_result.backend,
|
||
"planner_model": planning_result.model_name,
|
||
"planner_reason": planning_result.reason,
|
||
"planner_debug": planning_result.metadata,
|
||
"step_results": {},
|
||
"confirmed_steps": [],
|
||
},
|
||
)
|
||
|
||
def _continue_planned_workflow(
|
||
self,
|
||
session: SessionState,
|
||
workflow: Workflow,
|
||
routing_debug: RoutingDebug | None,
|
||
carry_messages: list[str] | None = None,
|
||
) -> ChatResponse:
|
||
step_messages: list[str] = list(carry_messages or [])
|
||
step_results = workflow.meta.setdefault("step_results", {})
|
||
confirmed_steps = workflow.meta.setdefault("confirmed_steps", [])
|
||
for step in workflow.steps:
|
||
if step.status in {"completed", "skipped"}:
|
||
continue
|
||
if step.depends_on and not all(self._is_step_completed(workflow, dependency) for dependency in step.depends_on):
|
||
continue
|
||
intent = self.intent_registry.get(step.intent_id)
|
||
merged_slots = session.slots.copy()
|
||
merged_slots.update(step.slots)
|
||
merged_slots = self._normalize_slots_for_intent(intent.intent_id, merged_slots)
|
||
step.slots = merged_slots.copy()
|
||
session.current_intent = intent.intent_id
|
||
missing_slots = self._pending_slots_for_intent(intent, merged_slots)
|
||
if missing_slots:
|
||
workflow.status = "waiting_slot"
|
||
session.pending_slots = missing_slots
|
||
session.status = "waiting_slot"
|
||
workflow.missing_slots = [
|
||
MissingSlot(
|
||
slot_name=slot,
|
||
ask_template=self._default_ask_template(intent, slot),
|
||
priority=index + 1,
|
||
)
|
||
for index, slot in enumerate(missing_slots)
|
||
]
|
||
session.workflow = workflow.model_dump()
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="ask_slot",
|
||
reply_text=self.response_policy.ask_for_slot(
|
||
intent,
|
||
missing_slots[0],
|
||
self._default_ask_template(intent, missing_slots[0]),
|
||
),
|
||
intent=intent.intent_id,
|
||
domain=intent.domain,
|
||
status=session.status,
|
||
pending_slots=missing_slots,
|
||
|
||
filled_slots=merged_slots,
|
||
workflow=workflow,
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
condition_state = self._evaluate_step_condition(step, workflow)
|
||
if condition_state is False:
|
||
step.status = "skipped"
|
||
reason = step.condition.get("description") if step.condition else None
|
||
step_messages.append(self.response_policy.step_skipped(intent, reason))
|
||
continue
|
||
|
||
if step.requires_confirmation and step.step_id not in confirmed_steps:
|
||
step.status = "waiting_confirmation"
|
||
workflow.status = "waiting_confirmation"
|
||
session.pending_slots = ["confirmation"]
|
||
session.status = "waiting_confirmation"
|
||
workflow.meta["pending_confirmation"] = {
|
||
"step_id": step.step_id,
|
||
"intent_id": intent.intent_id,
|
||
"detail": step.condition.get("description") if step.condition else None,
|
||
}
|
||
session.workflow = workflow.model_dump()
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="ask_confirmation",
|
||
reply_text=self.response_policy.ask_for_confirmation(
|
||
intent,
|
||
str(workflow.meta["pending_confirmation"].get("detail") or "").strip() or None,
|
||
),
|
||
intent=intent.intent_id,
|
||
domain=intent.domain,
|
||
status=session.status,
|
||
pending_slots=["confirmation"],
|
||
|
||
filled_slots=merged_slots,
|
||
workflow=workflow,
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
plugin_result = self.plugins.execute(intent.plugin_id, merged_slots)
|
||
step.status = "completed"
|
||
step_results[step.step_id] = plugin_result
|
||
session.slots.update(merged_slots)
|
||
self._update_context_memory(session, intent.intent_id, merged_slots)
|
||
step_messages.append(self.response_policy.workflow_result(intent, plugin_result))
|
||
|
||
workflow.status = "completed"
|
||
workflow.missing_slots = []
|
||
session.pending_slots = []
|
||
session.status = "completed"
|
||
session.current_intent = workflow.intent_id
|
||
session.workflow = workflow.model_dump()
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="workflow_result",
|
||
reply_text=self.response_policy.workflow_summary(step_messages),
|
||
intent=workflow.intent_id,
|
||
domain=self._domain_for_intent(workflow.intent_id),
|
||
status=session.status,
|
||
pending_slots=[],
|
||
|
||
filled_slots=session.slots,
|
||
workflow=workflow,
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
def _requires_confirmation(self, intent) -> bool:
|
||
return self.dialog_rules.requires_confirmation(intent.intent_id, intent.risk_level)
|
||
|
||
def _is_step_completed(self, workflow: Workflow, step_id: str) -> bool:
|
||
dependency = next((item for item in workflow.steps if item.step_id == step_id), None)
|
||
return dependency is not None and dependency.status == "completed"
|
||
|
||
def _evaluate_step_condition(self, step: WorkflowStep, workflow: Workflow) -> bool | None:
|
||
if not step.condition:
|
||
return None
|
||
source_step_id = self._resolve_condition_source_step(step, workflow)
|
||
if source_step_id is None:
|
||
return None
|
||
step_results = workflow.meta.get("step_results", {})
|
||
source_result = step_results.get(source_step_id, {})
|
||
if not isinstance(source_result, dict):
|
||
return None
|
||
field_name = str(step.condition.get("field", "")).strip()
|
||
operator = str(step.condition.get("operator", "equals")).strip() or "equals"
|
||
expected = step.condition.get("value")
|
||
actual = source_result.get("data", {}).get(field_name)
|
||
if operator == "equals":
|
||
return actual == expected
|
||
if operator == "not_equals":
|
||
return actual != expected
|
||
if operator == "in" and isinstance(expected, list):
|
||
return actual in expected
|
||
return None
|
||
|
||
def _resolve_condition_source_step(self, step: WorkflowStep, workflow: Workflow) -> str | None:
|
||
source_step = step.condition.get("source_step")
|
||
try:
|
||
if source_step is not None:
|
||
source_index = int(source_step)
|
||
matched = next((item for item in workflow.steps if item.step == source_index), None)
|
||
if matched is not None:
|
||
return matched.step_id
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if step.depends_on:
|
||
return step.depends_on[0]
|
||
return None
|
||
|
||
def _handle_confirmation(self, session: SessionState, text: str) -> ChatResponse:
|
||
workflow = self._workflow_from_session(session)
|
||
routing_debug = self._routing_debug_from_session(session)
|
||
if workflow is None:
|
||
return self._fallback_response(session.session_id, routing_debug)
|
||
|
||
pending = workflow.meta.get("pending_confirmation", {})
|
||
step_id = str(pending.get("step_id", "")).strip()
|
||
if not step_id:
|
||
return self._fallback_response(session.session_id, routing_debug)
|
||
|
||
decision = self._parse_confirmation_decision(text)
|
||
step = next((item for item in workflow.steps if item.step_id == step_id), None)
|
||
if step is None:
|
||
return self._fallback_response(session.session_id, routing_debug)
|
||
intent = self.intent_registry.get(step.intent_id)
|
||
|
||
if decision is None:
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="ask_confirmation",
|
||
reply_text=self.response_policy.confirm_retry(),
|
||
intent=intent.intent_id,
|
||
domain=intent.domain,
|
||
status=session.status,
|
||
pending_slots=["confirmation"],
|
||
|
||
filled_slots=session.slots,
|
||
workflow=workflow,
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|
||
|
||
workflow.meta.pop("pending_confirmation", None)
|
||
if decision is False:
|
||
step.status = "skipped"
|
||
session.pending_slots = []
|
||
session.status = "running"
|
||
skipped_reason = self.response_policy.confirm_cancelled()
|
||
step_messages = [skipped_reason]
|
||
response = self._continue_planned_workflow(session, workflow, routing_debug, carry_messages=step_messages)
|
||
return response
|
||
|
||
confirmed_steps = workflow.meta.setdefault("confirmed_steps", [])
|
||
if step_id not in confirmed_steps:
|
||
confirmed_steps.append(step_id)
|
||
session.pending_slots = []
|
||
session.status = "running"
|
||
return self._continue_planned_workflow(session, workflow, routing_debug)
|
||
|
||
def _workflow_from_session(self, session: SessionState) -> Workflow | None:
|
||
if session.workflow is None:
|
||
return None
|
||
return Workflow.model_validate(session.workflow)
|
||
|
||
def _attach_rewrite_debug(
|
||
self,
|
||
routing_debug: RoutingDebug,
|
||
rewrite_result: RewriteResult,
|
||
elapsed_ms: float | None = None,
|
||
) -> None:
|
||
routing_debug.stages.insert(
|
||
0,
|
||
MatcherStageDebug(
|
||
stage="rewrite",
|
||
accepted=rewrite_result.applied,
|
||
score=1.0 if rewrite_result.applied else 0.0,
|
||
elapsed_ms=elapsed_ms,
|
||
reason=rewrite_result.reason or "no rewrite needed",
|
||
model_name="context-rewrite",
|
||
metadata={
|
||
"original_text": rewrite_result.original_text,
|
||
"rewritten_text": rewrite_result.rewritten_text,
|
||
**rewrite_result.metadata,
|
||
},
|
||
),
|
||
)
|
||
|
||
def _update_context_memory(
|
||
self,
|
||
session: SessionState,
|
||
intent_id: str,
|
||
extracted_slots: dict[str, object],
|
||
) -> None:
|
||
session.context_memory["last_intent"] = intent_id
|
||
if "temperature" in extracted_slots:
|
||
session.context_memory["last_temperature"] = extracted_slots["temperature"]
|
||
if "destination" in extracted_slots:
|
||
session.context_memory["last_destination"] = extracted_slots["destination"]
|
||
if "song" in extracted_slots:
|
||
session.context_memory["last_song"] = extracted_slots["song"]
|
||
if "genre" in extracted_slots:
|
||
session.context_memory["last_genre"] = extracted_slots["genre"]
|
||
|
||
def _normalize_slots_for_intent(self, intent_id: str, slots: dict[str, object]) -> dict[str, object]:
|
||
normalized = dict(slots)
|
||
if intent_id == "cabin_set_ac":
|
||
temperature = self._normalize_temperature_value(normalized.get("temperature"))
|
||
if temperature is not None:
|
||
normalized["temperature"] = temperature
|
||
else:
|
||
normalized.pop("temperature", None)
|
||
if intent_id == "cabin_play_music":
|
||
media_query = str(normalized.get("media_query") or "").strip()
|
||
if media_query and "song" not in normalized and "genre" not in normalized:
|
||
if any(genre in media_query for genre in self._MUSIC_GENRE_HINTS):
|
||
matched = next((genre for genre in self._MUSIC_GENRE_HINTS if genre in media_query), None)
|
||
if matched:
|
||
normalized["genre"] = matched
|
||
else:
|
||
normalized["song"] = media_query
|
||
normalized.pop("media_query", None)
|
||
return normalized
|
||
|
||
def _continue_active_task(self, session: SessionState, text: str) -> ChatResponse | None:
|
||
if session.current_intent is None:
|
||
return None
|
||
if session.status not in {"waiting_slot", "waiting_confirmation"}:
|
||
return None
|
||
if session.status == "waiting_confirmation":
|
||
return None
|
||
intent = self.intent_registry.get(session.current_intent)
|
||
extracted_slots = self._extract_continuation_slots(session, intent, text)
|
||
if not extracted_slots:
|
||
return None
|
||
session.slots.update(extracted_slots)
|
||
self._update_context_memory(session, intent.intent_id, extracted_slots)
|
||
if session.routing_debug is not None:
|
||
routing_debug = RoutingDebug.model_validate(session.routing_debug)
|
||
routing_debug.extracted_slots = session.slots.copy()
|
||
session.routing_debug = routing_debug.model_dump()
|
||
existing_workflow = self._workflow_from_session(session)
|
||
if existing_workflow is not None and existing_workflow.workflow_type != "single":
|
||
return self._continue_planned_workflow(
|
||
session,
|
||
existing_workflow,
|
||
self._routing_debug_from_session(session),
|
||
)
|
||
return self._build_response_from_session(session)
|
||
|
||
def _extract_continuation_slots(
|
||
self,
|
||
session: SessionState,
|
||
intent,
|
||
text: str,
|
||
) -> dict[str, object]:
|
||
extracted = self.router.extract_slots(text, intent)
|
||
if extracted:
|
||
return extracted
|
||
normalized = text.strip()
|
||
if not normalized:
|
||
return {}
|
||
candidate_texts = self._continuation_candidate_texts(intent.intent_id, normalized, session.pending_slots)
|
||
merged: dict[str, object] = {}
|
||
for candidate_text in candidate_texts:
|
||
candidate_slots = self.router.extract_slots(candidate_text, intent)
|
||
if candidate_slots:
|
||
merged.update(candidate_slots)
|
||
break
|
||
if merged:
|
||
return merged
|
||
if intent.intent_id == "cabin_play_music" and self._requires_music_query(session.pending_slots):
|
||
compact = normalized.strip("，。,.!?！？ ")
|
||
if not compact:
|
||
return {}
|
||
if self.dialog_act_engine.detect(compact) in {"thanks", "greeting", "goodbye"}:
|
||
return {}
|
||
matched_genre = next((genre for genre in self._MUSIC_GENRE_HINTS if genre in compact), None)
|
||
if matched_genre is not None:
|
||
return {"genre": matched_genre}
|
||
if len(compact) <= 20:
|
||
return {"song": compact}
|
||
return {}
|
||
|
||
def _continuation_candidate_texts(
|
||
self,
|
||
intent_id: str,
|
||
text: str,
|
||
pending_slots: list[str],
|
||
) -> list[str]:
|
||
pending = list(pending_slots)
|
||
if intent_id == "cabin_play_music" and self._requires_music_query(pending):
|
||
return [f"播放{text}", f"来一首{text}", f"来点{text}"]
|
||
if intent_id == "cabin_nav_to" and "destination" in pending:
|
||
return [f"导航去{text}", f"带我去{text}"]
|
||
if intent_id == "cabin_set_ac" and "temperature" in pending:
|
||
return [f"把空调调到{text}", f"温度设成{text}"]
|
||
if intent_id in {"cs_query_order", "cs_query_logistics", "cs_cancel_order"} and "order_id" in pending:
|
||
prefix = {
|
||
"cs_query_order": "查一下订单",
|
||
"cs_query_logistics": "查一下物流",
|
||
"cs_cancel_order": "取消订单",
|
||
}.get(intent_id, "")
|
||
return [f"{prefix}{text}"] if prefix else [text]
|
||
return []
|
||
|
||
def _pending_slots_for_intent(self, intent, slots: dict[str, object]) -> list[str]:
|
||
pending_slots = [slot for slot in intent.required_slots if slot not in slots]
|
||
if intent.intent_id == "cabin_play_music" and not self._has_music_query(slots):
|
||
pending_slots.append("media_query")
|
||
deduped: list[str] = []
|
||
for slot in pending_slots:
|
||
if slot not in deduped:
|
||
deduped.append(slot)
|
||
return deduped
|
||
|
||
def _default_ask_template(self, intent, slot_name: str) -> str:
|
||
if slot_name in intent.ask_templates:
|
||
return str(intent.ask_templates[slot_name])
|
||
if slot_name == "media_query":
|
||
return "想听什么风格或者具体的歌名？"
|
||
return "请补充一个关键信息。"
|
||
|
||
def _has_music_query(self, slots: dict[str, object]) -> bool:
|
||
return bool(str(slots.get("song") or "").strip() or str(slots.get("genre") or "").strip())
|
||
|
||
def _requires_music_query(self, pending_slots: list[str]) -> bool:
|
||
return "media_query" in pending_slots or not pending_slots
|
||
|
||
def _looks_like_parallel_request(self, text: str, fusion_stage: MatcherStageDebug | None) -> bool:
|
||
_ = fusion_stage
|
||
compact = re.sub(r"\s+", "", text)
|
||
return bool(re.search(r"(和|跟|以及|外加).+", compact) and len(compact) >= 5)
|
||
|
||
def _normalize_temperature_value(self, raw_value: object) -> int | None:
|
||
if isinstance(raw_value, bool):
|
||
return None
|
||
if isinstance(raw_value, (int, float)):
|
||
value = int(raw_value)
|
||
elif isinstance(raw_value, str):
|
||
match = re.search(r"-?\d+", raw_value)
|
||
if not match:
|
||
return None
|
||
value = int(match.group(0))
|
||
else:
|
||
return None
|
||
return max(16, min(30, value))
|
||
|
||
def _record_turn(
|
||
self,
|
||
session: SessionState,
|
||
user_text: str,
|
||
agent_text: str,
|
||
rewrite_result: RewriteResult,
|
||
) -> None:
|
||
session.last_user_text = user_text
|
||
session.last_agent_text = agent_text
|
||
session.context_memory["last_raw_user_text"] = user_text
|
||
session.context_memory["last_rewritten_text"] = rewrite_result.rewritten_text
|
||
|
||
def _update_dialog_act(self, session: SessionState, text: str) -> None:
|
||
session.context_memory["last_dialog_act"] = self.dialog_act_engine.detect(text)
|
||
|
||
def _has_active_task(self, session: SessionState) -> bool:
|
||
if session.status in {"waiting_slot", "waiting_confirmation", "running", "understanding"}:
|
||
return True
|
||
workflow = self._workflow_from_session(session)
|
||
if workflow is not None and workflow.status not in {"completed", "failed"}:
|
||
return True
|
||
return False
|
||
|
||
def _is_stop_request(self, text: str) -> bool:
|
||
return self.dialog_rules.is_stop_request(text)
|
||
|
||
def _reset_active_task(self, session: SessionState) -> None:
|
||
session.status = "idle"
|
||
session.current_intent = None
|
||
session.pending_slots = []
|
||
session.slots = {}
|
||
session.workflow = None
|
||
session.routing_debug = None
|
||
|
||
def _parse_confirmation_decision(self, text: str) -> bool | None:
|
||
return self.dialog_rules.parse_confirmation_decision(text)
|
||
|
||
def _try_knowledge_llm(
|
||
self,
|
||
session: SessionState,
|
||
user_text: str,
|
||
routing_debug: RoutingDebug,
|
||
rewrite_result: RewriteResult,
|
||
) -> ChatResponse | None:
|
||
"""BERT 未命中时尝试 LLM + knowledge_search function call。
|
||
有结果返回 ChatResponse（含 knowledge_content），无结果返回 None。
|
||
"""
|
||
if self.knowledge_llm is None:
|
||
return None
|
||
|
||
result = self.knowledge_llm.reply(user_text)
|
||
|
||
# LLM 调用失败或本地兜底但没有文档 → 退回原始 fallback
|
||
if result.backend == "local-fallback" and result.doc_id is None:
|
||
return None
|
||
|
||
return ChatResponse(
|
||
session_id=session.session_id,
|
||
reply_type="text",
|
||
reply_text=result.reply_text,
|
||
intent=None,
|
||
domain="knowledge",
|
||
decision="knowledge_llm",
|
||
decision_reason=(
|
||
f"BERT 未命中，LLM 通过 knowledge_search 命中文档：{result.doc_id}"
|
||
if result.doc_id
|
||
else "BERT 未命中，LLM 自主回答"
|
||
),
|
||
status="completed",
|
||
knowledge_doc_id=result.doc_id,
|
||
knowledge_doc_title=result.doc_title,
|
||
knowledge_content=result.doc_content,
|
||
routing_debug=routing_debug,
|
||
trace_id=self._trace_id(),
|
||
)
|