diff --git a/backend/app/api/docs/chatbot/publish_flow.md b/backend/app/api/docs/chatbot/publish_flow.md index 29079d486..fd924d8e0 100644 --- a/backend/app/api/docs/chatbot/publish_flow.md +++ b/backend/app/api/docs/chatbot/publish_flow.md @@ -1,5 +1,5 @@ Publish the chatbot's draft so new sessions run on it. Returns the now-live version. -The draft is validated strictly: node ids are unique; at most one `input` and one `output` node; every edge's `source` and `target` is a node in the flow; every agent node has `data.config_id` and a non-empty `data.prompt`, and its config resolves to an Anthropic text config in this project with a model set (and a valid thinking budget). The project must have Anthropic credentials configured, since agent nodes call Anthropic with the project's own key. Only `onboarding` agent nodes can run for now — `coordinator` and `rag` nodes are rejected as not supported yet. The flow needs an entry node: with no edges it must contain exactly one agent node; otherwise the `input` node must have exactly one outgoing edge, leading to an agent node. A node without `data.config_version` gets the config's latest version written into the stored node, so later config edits don't change the published flow. +The draft is validated strictly: node ids are unique; at most one `input` and one `output` node; every edge's `source` and `target` is a node in the flow; every agent node has `data.config_id` and a non-empty `data.prompt`, and its config resolves to an Anthropic text config in this project with a model set (and a valid thinking budget). The project must have Anthropic credentials configured, since agent nodes call Anthropic with the project's own key. `onboarding` and `coordinator` agent nodes can run; `rag` nodes are rejected as not supported yet. A flow has at most one `coordinator`; it needs at least one outgoing edge, every outgoing edge must lead to an `onboarding` node, and each such edge needs a non-empty `data.intent` that is unique among that coordinator's edges (`data.description` is optional and helps the routing). The flow needs an entry node: with no edges it must contain exactly one agent node; otherwise the `input` node must have exactly one outgoing edge, leading to an agent node. A node without `data.config_version` gets the config's latest version written into the stored node, so later config edits don't change the published flow. The previous live version (if any) stops being live in the same transaction. Sessions already running stay pinned to the version they started on. Returns 404 for an unknown chatbot, 409 if there is no draft to publish (save the flow first) or on a concurrent change, and 422 listing every validation problem. diff --git a/backend/app/api/docs/chatbot/save_flow.md b/backend/app/api/docs/chatbot/save_flow.md index d940014cb..60cf6f53e 100644 --- a/backend/app/api/docs/chatbot/save_flow.md +++ b/backend/app/api/docs/chatbot/save_flow.md @@ -4,4 +4,4 @@ The **first** save of a chatbot creates version 1 and makes it live immediately, Every later save is a draft that needs a publish to go live: if the highest version is not live it is overwritten, otherwise a new version `max + 1` is created with `is_live: false`. Live versions are never modified, so editing never affects running conversations. -Draft saves check only the shape, so half-built flows save: each node needs `id` (1–255 chars), `type` (`input`, `output`, `coordinator`, `onboarding`, `rag`), `position {x, y}` and an object `data`; agent nodes (`coordinator`, `onboarding`, `rag`) may carry `data.config_id` (UUID), `data.config_version` (≥ 1) and `data.prompt`. Each edge needs `id`, `source`, `target`; `sourceHandle`, `targetHandle` and `data` (e.g. `intent`, `description`) are optional. Extra keys are stored and returned as sent. Returns 404 for an unknown chatbot, 409 if the flow was changed concurrently (reload and retry), and 422 for an invalid first save. +Draft saves check only the shape, so half-built flows save: each node needs `id` (1–255 chars), `type` (`input`, `output`, `coordinator`, `onboarding`, `rag`), `position {x, y}` and an object `data`; agent nodes (`coordinator`, `onboarding`, `rag`) may carry `data.config_id` (UUID), `data.config_version` (≥ 1) and `data.prompt`. Each edge needs `id`, `source`, `target`; `sourceHandle`, `targetHandle` and `data` (e.g. `intent`, `description`) are optional; a coordinator routes on its outgoing edges' `data.intent` / `data.description`, and the intent rules are enforced only at publish. Extra keys are stored and returned as sent. Returns 404 for an unknown chatbot, 409 if the flow was changed concurrently (reload and retry), and 422 for an invalid first save. diff --git a/backend/app/api/docs/chatbot/send_session_message.md b/backend/app/api/docs/chatbot/send_session_message.md index e6d1034d0..95b72e967 100644 --- a/backend/app/api/docs/chatbot/send_session_message.md +++ b/backend/app/api/docs/chatbot/send_session_message.md @@ -4,4 +4,6 @@ The user `message` is stored immediately (tagged with the session's active node) The turn runs the session's active node of its pinned flow version: the node's `data.prompt` is the system prompt and its `data.config_id` / `data.config_version` supply the model. Each turn re-reads the conversation against that prompt and saves every field the user has given into `collected_fields`, a flat `{field_name: value}` map with snake_case keys (values are only ever taken from what the user said; a correction overwrites the old value). `completion_status` is `complete` once every required field is collected and `in_progress` otherwise; it is computed from a structured extraction, not from the reply text. A `complete` session still accepts messages. `status` (`active`/`inactive`) is separate and says whether the session is in use: once a newer session is created for the same contact and chatbot, or the chatbot is deleted, this one is `inactive` and rejects new messages with 409. A turn that was already queued when that happened is not run; Kaapi POSTs the failure payload with error `[KAAPI] Chatbot session was superseded by a newer session; start a new session.` instead. +When the active node is a `coordinator`, it classifies the message against its outgoing edges' intents. On a clear match the session moves to that `onboarding` node, which writes the reply in the same turn (one webhook). If nothing clearly matches, the coordinator replies with a clarifying question and the session stays on it; the opening greeting is always the coordinator's own. `node_id` in the webhook is the node that produced the reply. When a routed agent finishes (`completion_status: complete`), the session goes back to the coordinator, so the next message is routed afresh and `completion_status` returns to `in_progress`. In coordinator flows, treat `complete` as "this sub-flow is done" for that webhook only, not as the end of the session. `collected_fields` is one map shared by every agent in the session. + On the first call of a session, omit `message` to get the bot's opening greeting; after that `message` is required. `webhook_url` must be a public HTTPS URL. Returns 404 if the session isn't in the caller's project, 409 if the session is inactive, and 422 for an invalid `webhook_url` or a missing `message` on a started session. diff --git a/backend/app/crud/chatbot.py b/backend/app/crud/chatbot.py index 7ab7e93b2..b924c539d 100644 --- a/backend/app/crud/chatbot.py +++ b/backend/app/crud/chatbot.py @@ -367,18 +367,21 @@ def update_chatbot_session_progress( session_id: UUID, collected_fields: dict[str, Any], completion_status: ChatbotCompletionStatusEnum, + active_node_id: str | None = None, ) -> ChatbotSession | None: + """`active_node_id=None` leaves the session's current node unchanged.""" chatbot_session = session.get(ChatbotSession, session_id) if chatbot_session is None: return None - chatbot_session.sqlmodel_update( - { - "collected_fields": collected_fields, - "completion_status": completion_status, - "updated_at": now(), - } - ) + updates: dict[str, Any] = { + "collected_fields": collected_fields, + "completion_status": completion_status, + "updated_at": now(), + } + if active_node_id is not None: + updates["active_node_id"] = active_node_id + chatbot_session.sqlmodel_update(updates) session.add(chatbot_session) session.commit() session.refresh(chatbot_session) diff --git a/backend/app/services/chatbot/agents/coordinator/__init__.py b/backend/app/services/chatbot/agents/coordinator/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/services/chatbot/agents/coordinator/graph.py b/backend/app/services/chatbot/agents/coordinator/graph.py new file mode 100644 index 000000000..7c145344a --- /dev/null +++ b/backend/app/services/chatbot/agents/coordinator/graph.py @@ -0,0 +1,166 @@ +"""Coordinator turn: one structured-output call that routes the user to an agent or asks to clarify. + +No LangGraph here — a single model call doesn't need a graph. +""" + +import asyncio +import logging +from dataclasses import dataclass +from typing import NamedTuple + +import anthropic +from langchain_core.exceptions import OutputParserException +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.messages import BaseMessage +from pydantic import BaseModel, Field, ValidationError + +from app.services.chatbot.agents.coordinator.prompts import ( + FALLBACK_CLARIFYING_QUESTION, + ROUTES_TEMPLATE, + ROUTING_INSTRUCTION, +) +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.services.chatbot.history import summarize_if_needed, to_langchain_messages +from app.services.chatbot.types import ChatbotMessageDict +from app.services.chatbot.utils import ( + STRUCTURED_OUTPUT_METHOD, + ChatbotModelSettings, + get_chat_model, + to_provider_error, +) + +logger = logging.getLogger(__name__) + + +class RouteDecision(BaseModel): + intent: str | None = Field( + default=None, + description="Exactly one intent name from the list, or null when none clearly matches.", + ) + clarifying_question: str | None = Field( + default=None, + description="Question to ask the user when intent is null; otherwise null.", + ) + + +class CoordinatorRoute(NamedTuple): + intent: str + description: str | None + target_node_id: str + + +@dataclass(frozen=True) +class CoordinatorDecision: + """Exactly one of `target_node_id` / `clarifying_question` is set.""" + + target_node_id: str | None + clarifying_question: str | None + + +def _build_routing_prompt(*, system_prompt: str, routes: list[CoordinatorRoute]) -> str: + route_lines = [ + f"- {route.intent}" + (f": {route.description}" if route.description else "") + for route in routes + ] + return ( + system_prompt + + ROUTING_INSTRUCTION + + ROUTES_TEMPLATE.format(routes="\n".join(route_lines)) + ) + + +def _decide_route( + chat_model: BaseChatModel, + messages: list[BaseMessage], + *, + history_token_limit: int, +) -> RouteDecision: + conversation_view = summarize_if_needed( + chat_model, messages, history_token_limit=history_token_limit + ) + routing_model = chat_model.with_structured_output( + RouteDecision, method=STRUCTURED_OUTPUT_METHOD + ) + # Both are ValueError subclasses; mapped here so the caller's ValueError + # catch doesn't mislabel them as a model-configuration problem. + try: + decision = routing_model.invoke(conversation_view) + except (OutputParserException, ValidationError) as e: + error_message = ( + "[ANTHROPIC] Coordinator routing returned malformed structured output — " + f"retry the turn. If persistent, contact Kaapi. (code: {type(e).__name__})" + ) + logger.warning(f"[_decide_route] {error_message}", exc_info=True) + raise ChatbotProviderError(error_message) from e + + if not isinstance(decision, RouteDecision): + error_message = ( + "[ANTHROPIC] Coordinator routing returned an unexpected output shape — " + "retry the turn. If persistent, contact Kaapi. (code: UnexpectedOutputType)" + ) + logger.warning(f"[_decide_route] {error_message}") + raise ChatbotProviderError(error_message) + return decision + + +def _match_route( + intent: str | None, routes: list[CoordinatorRoute] +) -> CoordinatorRoute | None: + if intent is None: + return None + wanted = intent.strip().casefold() + return next((route for route in routes if route.intent.casefold() == wanted), None) + + +async def run_coordinator_turn( + *, + system_prompt: str, + history: list[ChatbotMessageDict], + user_message: str | None, + routes: list[CoordinatorRoute], + model_settings: ChatbotModelSettings, +) -> CoordinatorDecision: + """Classify the latest user message against `routes`, or return a clarifying question.""" + if user_message is None and history: + raise ChatbotInputError( + "[KAAPI] `message` is required once `history` is non-empty; omit it only " + "on the first call to get the opening greeting." + ) + + messages = to_langchain_messages( + system_prompt=_build_routing_prompt(system_prompt=system_prompt, routes=routes), + history=history, + user_message=user_message, + ) + try: + chat_model = get_chat_model(model_settings) + # ChatAnthropic.invoke is sync; keep it off the event loop. + decision = await asyncio.to_thread( + _decide_route, + chat_model, + messages, + history_token_limit=model_settings.history_token_limit, + ) + except (anthropic.APIConnectionError, anthropic.APIStatusError, ValueError) as e: + raise to_provider_error(e, model=model_settings.model) from e + + # The opening turn has only the synthetic kickoff message, so it must never hand off. + if user_message is not None: + route = _match_route(decision.intent, routes) + if route is not None: + logger.info( + f"[run_coordinator_turn] Routed | intent: {route.intent}, " + f"target_node_id: {route.target_node_id}" + ) + return CoordinatorDecision( + target_node_id=route.target_node_id, clarifying_question=None + ) + + question = decision.clarifying_question + if question is None or not question.strip(): + question = FALLBACK_CLARIFYING_QUESTION + logger.info( + f"[run_coordinator_turn] Clarifying | model_intent: {decision.intent}, " + f"opening_turn: {user_message is None}" + ) + return CoordinatorDecision(target_node_id=None, clarifying_question=question) diff --git a/backend/app/services/chatbot/agents/coordinator/prompts.py b/backend/app/services/chatbot/agents/coordinator/prompts.py new file mode 100644 index 000000000..735005adb --- /dev/null +++ b/backend/app/services/chatbot/agents/coordinator/prompts.py @@ -0,0 +1,14 @@ +ROUTING_INSTRUCTION = ( + "\n\nYou route the user to the right assistant. Read the conversation and decide " + "which of the intents below the user's latest message expresses.\n" + "1. intent: copy exactly one intent name from the list when the user's need clearly " + "matches it; otherwise null. Never invent an intent that is not listed.\n" + "2. clarifying_question: when intent is null, a short, friendly question in the " + "user's language that helps them say which of the listed options they need; " + "otherwise null." +) +ROUTES_TEMPLATE = "\n\n\n{routes}\n" + +FALLBACK_CLARIFYING_QUESTION = ( + "Could you tell me a bit more about what you need help with?" +) diff --git a/backend/app/services/chatbot/agents/onboarding/graph.py b/backend/app/services/chatbot/agents/onboarding/graph.py index 4c8f32ded..cf4c968e5 100644 --- a/backend/app/services/chatbot/agents/onboarding/graph.py +++ b/backend/app/services/chatbot/agents/onboarding/graph.py @@ -17,7 +17,6 @@ from app.services.chatbot.agents.onboarding.prompts import ( EXTRACTION_FLOW_PROMPT_TEMPLATE, EXTRACTION_INSTRUCTION, - EXTRACTION_METHOD, EXTRACTION_PRIOR_FIELDS_TEMPLATE, PROGRESS_COLLECTED_TEMPLATE, PROGRESS_COMPLETE_NOTE, @@ -37,7 +36,12 @@ to_langchain_messages, ) from app.services.chatbot.types import ChatbotMessageDict, ChatbotTurnResult -from app.services.chatbot.utils import ChatbotModelSettings, get_chat_model +from app.services.chatbot.utils import ( + STRUCTURED_OUTPUT_METHOD, + ChatbotModelSettings, + get_chat_model, + to_provider_error, +) ExtractionModel = Runnable[LanguageModelInput, dict | BaseModel] @@ -163,28 +167,8 @@ def chatbot_node(state: OnboardingState) -> dict[str, list[BaseMessage]]: def _get_extraction_model(chat_model: BaseChatModel) -> ExtractionModel: - return chat_model.with_structured_output(FieldExtraction, method=EXTRACTION_METHOD) - - -def _to_provider_error(error: Exception, *, model: str) -> ChatbotProviderError: - if isinstance(error, anthropic.APITimeoutError): - return ChatbotProviderError( - "[KAAPI] Request timed out — retry smaller. If persistent, contact Kaapi. " - "(code: APITimeoutError)" - ) - if isinstance(error, anthropic.APIConnectionError): - return ChatbotProviderError( - "[KAAPI] Network/DNS issue reaching provider — check connectivity. " - "If persistent, contact Kaapi. (code: APIConnectionError)" - ) - if isinstance(error, anthropic.APIStatusError): - return ChatbotProviderError( - f"[ANTHROPIC] Chat model call failed: {error.message} " - f"(code: {error.status_code}, request_id: {error.request_id})" - ) - return ChatbotProviderError( - f"[KAAPI] Unsupported model configuration for '{model}': {error} " - "— update the config's model, thinking, or effort. (code: ValueError)" + return chat_model.with_structured_output( + FieldExtraction, method=STRUCTURED_OUTPUT_METHOD ) @@ -223,7 +207,7 @@ async def run_chatbot_turn( # ChatAnthropic.invoke is sync; keep it off the event loop. result = await asyncio.to_thread(graph.invoke, initial_state) except (anthropic.APIConnectionError, anthropic.APIStatusError, ValueError) as e: - raise _to_provider_error(e, model=model_settings.model) from e + raise to_provider_error(e, model=model_settings.model) from e return ChatbotTurnResult( reply=extract_final_reply(result["messages"]), diff --git a/backend/app/services/chatbot/agents/onboarding/prompts.py b/backend/app/services/chatbot/agents/onboarding/prompts.py index b483cbe1d..26e7e02b9 100644 --- a/backend/app/services/chatbot/agents/onboarding/prompts.py +++ b/backend/app/services/chatbot/agents/onboarding/prompts.py @@ -1,5 +1,3 @@ -EXTRACTION_METHOD = "json_schema" - EXTRACTION_INSTRUCTION = ( "You extract data from a chatbot conversation. You do not talk to the user.\n" "The admin flow prompt below defines the goal, the fields to collect, and the rules.\n" diff --git a/backend/app/services/chatbot/flow.py b/backend/app/services/chatbot/flow.py index b5f78bac9..b034d0ef0 100644 --- a/backend/app/services/chatbot/flow.py +++ b/backend/app/services/chatbot/flow.py @@ -41,10 +41,14 @@ FIRST_FLOW_VERSION = 1 MAX_INPUT_NODES = 1 MAX_OUTPUT_NODES = 1 -# The runtime only executes onboarding agents on this branch; coordinator/rag are stored but not publishable. +# rag nodes are stored but not publishable until the runtime can execute them. SUPPORTED_AGENT_NODE_TYPES: frozenset[ChatbotFlowNodeTypeEnum] = frozenset( - {ChatbotFlowNodeTypeEnum.ONBOARDING} + {ChatbotFlowNodeTypeEnum.ONBOARDING, ChatbotFlowNodeTypeEnum.COORDINATOR} ) +# One coordinator keeps "which coordinator does a finished agent return to" unambiguous. +MAX_COORDINATOR_NODES = 1 +EDGE_INTENT_KEY = "intent" +EDGE_DESCRIPTION_KEY = "description" _SUPPORTED_AGENT_NODE_TYPES_LABEL = ", ".join( sorted(node_type.value for node_type in SUPPORTED_AGENT_NODE_TYPES) ) @@ -206,6 +210,63 @@ def _list_structural_problems( return problems +def get_edge_intent(edge: ChatbotFlowEdge) -> str | None: + """The edge's stripped `data.intent`, or None when absent/blank/not a string.""" + intent = edge.data.get(EDGE_INTENT_KEY) + if not isinstance(intent, str) or not intent.strip(): + return None + return intent.strip() + + +def _list_coordinator_problems( + *, nodes: list[ChatbotFlowNode], edges: list[ChatbotFlowEdge] +) -> list[str]: + problems: list[str] = [] + + coordinator_nodes = [ + node for node in nodes if node.type == ChatbotFlowNodeTypeEnum.COORDINATOR + ] + if len(coordinator_nodes) > MAX_COORDINATOR_NODES: + problems.append( + f"at most {MAX_COORDINATOR_NODES} coordinator node allowed " + f"(found {len(coordinator_nodes)})" + ) + + for coordinator in coordinator_nodes: + outgoing_edges = [edge for edge in edges if edge.source == coordinator.id] + if not outgoing_edges: + problems.append( + f"coordinator node '{coordinator.id}' needs at least one outgoing edge" + ) + continue + + seen_intents: set[str] = set() + for edge in outgoing_edges: + target = find_flow_node(nodes, edge.target) + # A missing target is already reported by the structural check. + if target is not None and target.type != ChatbotFlowNodeTypeEnum.ONBOARDING: + problems.append( + f"edge '{edge.id}' from coordinator '{coordinator.id}' must target " + f"an onboarding node (target '{target.id}' is '{target.type.value}')" + ) + + intent = get_edge_intent(edge) + if intent is None: + problems.append( + f"edge '{edge.id}' from coordinator '{coordinator.id}' is missing " + "data.intent" + ) + continue + # Casefolded to match runtime routing, which compares intents case-insensitively. + intent_key = intent.casefold() + if intent_key in seen_intents: + problems.append( + f"coordinator '{coordinator.id}' has duplicate intent '{intent}'" + ) + seen_intents.add(intent_key) + return problems + + def _pin_agent_node_config( *, session: Session, project_id: int, node: ChatbotFlowNode ) -> tuple[ChatbotFlowNode, list[str]]: @@ -259,6 +320,7 @@ def validate_flow_for_publish( ) -> list[ChatbotFlowNode]: """Strict publish-time checks; returns the nodes with every agent config_version pinned.""" problems = _list_structural_problems(nodes=nodes, edges=edges) + problems.extend(_list_coordinator_problems(nodes=nodes, edges=edges)) # Agent nodes call Anthropic with the project's own key, so a live flow needs one. if any(node.type in AGENT_NODE_TYPES for node in nodes): diff --git a/backend/app/services/chatbot/session.py b/backend/app/services/chatbot/session.py index bdf6370e4..255128c3b 100644 --- a/backend/app/services/chatbot/session.py +++ b/backend/app/services/chatbot/session.py @@ -1,5 +1,6 @@ import asyncio -from typing import NamedTuple +import logging +from typing import Any, NamedTuple from uuid import UUID from fastapi import HTTPException @@ -26,6 +27,13 @@ ChatbotSessionStatusEnum, ChatbotSessionSummary, ) +from app.services.chatbot.agents.coordinator.graph import ( + CoordinatorRoute, + run_coordinator_turn, +) +from app.services.chatbot.agents.coordinator.prompts import ( + FALLBACK_CLARIFYING_QUESTION, +) from app.services.chatbot.agents.onboarding.graph import run_chatbot_turn from app.services.chatbot.config import resolve_chatbot_model_settings from app.services.chatbot.exceptions import ( @@ -34,27 +42,50 @@ ChatbotProviderError, ) from app.services.chatbot.flow import ( + EDGE_DESCRIPTION_KEY, + EDGE_INTENT_KEY, + SUPPORTED_AGENT_NODE_TYPES, find_flow_node, get_chatbot_or_raise, parse_flow_edges, parse_flow_nodes, resolve_entry_node, ) -from app.services.chatbot.types import ChatbotMessageDict +from app.services.chatbot.types import ChatbotMessageDict, ChatbotTurnResult from app.services.chatbot.utils import ChatbotModelSettings from app.utils import APIResponse, get_webhook_secret, send_callback +logger = logging.getLogger(__name__) + SUPERSEDED_SESSION_ERROR = ( "[KAAPI] Chatbot session was superseded by a newer session; start a new session." ) NO_LIVE_FLOW_MESSAGE = "[KAAPI] Chatbot has no published flow; publish the flow first." +_RUNNABLE_NODE_TYPES_LABEL = ", ".join( + sorted(f"'{node_type.value}'" for node_type in SUPPORTED_AGENT_NODE_TYPES) +) +ACTIVE_NODE_LABEL = "Active node" +ROUTE_TARGET_LABEL = "Route target" + + class _TurnNode(NamedTuple): node_id: str prompt: str config_id: UUID config_version: int | None + node_type: ChatbotFlowNodeTypeEnum + + +_RouteTargets = dict[str, tuple[_TurnNode, ChatbotModelSettings]] + + +class _TurnOutcome(NamedTuple): + result: ChatbotTurnResult + reply_node_id: str + # None leaves the session's active node unchanged. + next_active_node_id: str | None def start_chatbot_session( @@ -134,33 +165,35 @@ def get_latest_chatbot_session_summary( def _resolve_turn_node( - *, flow_version: ChatbotFlowVersion | None, active_node_id: str + *, + flow_version: ChatbotFlowVersion, + active_node_id: str, + label: str = ACTIVE_NODE_LABEL, ) -> _TurnNode: - """The runnable onboarding node the session is on; ChatbotInputError when it can't run.""" - active_node = None - if flow_version is not None: - active_node = find_flow_node( - parse_flow_nodes(flow_version.nodes), active_node_id - ) + """The runnable agent node `active_node_id` names; ChatbotInputError when it can't run. + + `label` names the node's role in error messages (the active node or a coordinator's route target). + """ + active_node = find_flow_node(parse_flow_nodes(flow_version.nodes), active_node_id) if active_node is None: error_message = ( - f"[KAAPI] Active node '{active_node_id}' is not in the session's flow " + f"[KAAPI] {label} '{active_node_id}' is not in the session's flow " "version; start a new session." ) raise ChatbotInputError(error_message) - if active_node.type != ChatbotFlowNodeTypeEnum.ONBOARDING: + if active_node.type not in SUPPORTED_AGENT_NODE_TYPES: error_message = ( - f"[KAAPI] Active node '{active_node_id}' has type '{active_node.type.value}', " + f"[KAAPI] {label} '{active_node_id}' has type '{active_node.type.value}', " f"which the chatbot runtime does not support yet; only " - f"'{ChatbotFlowNodeTypeEnum.ONBOARDING.value}' nodes can run." + f"{_RUNNABLE_NODE_TYPES_LABEL} nodes can run." ) raise ChatbotInputError(error_message) node_data = active_node.agent_data() if node_data.prompt is None or node_data.config_id is None: error_message = ( - f"[KAAPI] Active node '{active_node_id}' is missing data.prompt or " + f"[KAAPI] {label} '{active_node_id}' is missing data.prompt or " "data.config_id; republish the flow and start a new session." ) raise ChatbotInputError(error_message) @@ -170,6 +203,123 @@ def _resolve_turn_node( prompt=node_data.prompt, config_id=node_data.config_id, config_version=node_data.config_version, + node_type=active_node.type, + ) + + +def _resolve_coordinator_routes( + *, + session: Session, + flow_version: ChatbotFlowVersion, + coordinator_node_id: str, + organization_id: int, + project_id: int, +) -> tuple[list[CoordinatorRoute], _RouteTargets]: + """Every outgoing route of the coordinator, plus each target's node and model settings. + + Intents and target types are enforced at publish, the only way a flow goes live. + """ + routes: list[CoordinatorRoute] = [] + targets: _RouteTargets = {} + for edge in parse_flow_edges(flow_version.edges): + if edge.source != coordinator_node_id: + continue + + target_node = _resolve_turn_node( + flow_version=flow_version, + active_node_id=edge.target, + label=ROUTE_TARGET_LABEL, + ) + description = edge.data.get(EDGE_DESCRIPTION_KEY) + targets[target_node.node_id] = ( + target_node, + resolve_chatbot_model_settings( + session=session, + organization_id=organization_id, + project_id=project_id, + config_id=target_node.config_id, + config_version=target_node.config_version, + max_tokens=None, + history_token_limit=None, + ), + ) + routes.append( + CoordinatorRoute( + intent=edge.data[EDGE_INTENT_KEY].strip(), + description=description if isinstance(description, str) else None, + target_node_id=target_node.node_id, + ) + ) + + if not routes: + error_message = ( + f"[KAAPI] Coordinator '{coordinator_node_id}' has no outgoing routes; " + "republish the flow and start a new session." + ) + raise ChatbotInputError(error_message) + return routes, targets + + +def _find_return_coordinator( + *, flow_version: ChatbotFlowVersion, node_id: str +) -> str | None: + """The coordinator that routes to `node_id`, or None when the node isn't behind one.""" + nodes = parse_flow_nodes(flow_version.nodes) + for edge in parse_flow_edges(flow_version.edges): + if edge.target != node_id: + continue + source_node = find_flow_node(nodes, edge.source) + if ( + source_node is not None + and source_node.type == ChatbotFlowNodeTypeEnum.COORDINATOR + ): + return source_node.id + return None + + +async def _run_coordinator_turn_and_handoff( + *, + coordinator_node: _TurnNode, + model_settings: ChatbotModelSettings, + routes: list[CoordinatorRoute], + targets: _RouteTargets, + history: list[ChatbotMessageDict], + user_message: str | None, + collected_fields: dict[str, Any], +) -> _TurnOutcome: + """Route, then let the chosen agent reply in the same turn (silent handoff).""" + decision = await run_coordinator_turn( + system_prompt=coordinator_node.prompt, + history=history, + user_message=user_message, + routes=routes, + model_settings=model_settings, + ) + if decision.target_node_id is None: + question = decision.clarifying_question or FALLBACK_CLARIFYING_QUESTION + return _TurnOutcome( + result=ChatbotTurnResult( + reply=question, collected_fields=collected_fields, is_complete=False + ), + reply_node_id=coordinator_node.node_id, + next_active_node_id=None, + ) + + target_node, target_settings = targets[decision.target_node_id] + turn_result = await run_chatbot_turn( + system_prompt=target_node.prompt, + history=history, + user_message=user_message, + model_settings=target_settings, + collected_fields=collected_fields, + ) + # A finished sub-flow hands the session back so the next message is routed afresh. + return _TurnOutcome( + result=turn_result, + reply_node_id=target_node.node_id, + next_active_node_id=( + coordinator_node.node_id if turn_result.is_complete else target_node.node_id + ), ) @@ -251,7 +401,17 @@ def process_chatbot_session_message( session, flow_version_id=chatbot_session.flow_version_id ) + # Everything the turn needs (incl. every route target's settings) is resolved + # here, so no DB access happens once the LLM calls start. + routes: list[CoordinatorRoute] = [] + targets: _RouteTargets = {} + return_node_id: str | None = None try: + if flow_version is None: + raise ChatbotInputError( + f"[KAAPI] {ACTIVE_NODE_LABEL} '{chatbot_session.active_node_id}' " + "is not in the session's flow version; start a new session." + ) turn_node = _resolve_turn_node( flow_version=flow_version, active_node_id=chatbot_session.active_node_id, @@ -265,6 +425,18 @@ def process_chatbot_session_message( max_tokens=None, history_token_limit=None, ) + if turn_node.node_type == ChatbotFlowNodeTypeEnum.COORDINATOR: + routes, targets = _resolve_coordinator_routes( + session=session, + flow_version=flow_version, + coordinator_node_id=turn_node.node_id, + organization_id=organization_id, + project_id=project_id, + ) + else: + return_node_id = _find_return_coordinator( + flow_version=flow_version, node_id=turn_node.node_id + ) except (HTTPException, ChatbotInputError) as e: error = str(e.detail) if isinstance(e, HTTPException) else str(e) _send_failure_webhook( @@ -278,15 +450,33 @@ def process_chatbot_session_message( history, user_message = _split_pending_turn(stored_messages) try: - turn_result = asyncio.run( - run_chatbot_turn( - system_prompt=turn_node.prompt, - history=history, - user_message=user_message, - model_settings=model_settings, - collected_fields=collected_fields, + if turn_node.node_type == ChatbotFlowNodeTypeEnum.COORDINATOR: + outcome = asyncio.run( + _run_coordinator_turn_and_handoff( + coordinator_node=turn_node, + model_settings=model_settings, + routes=routes, + targets=targets, + history=history, + user_message=user_message, + collected_fields=collected_fields, + ) + ) + else: + turn_result = asyncio.run( + run_chatbot_turn( + system_prompt=turn_node.prompt, + history=history, + user_message=user_message, + model_settings=model_settings, + collected_fields=collected_fields, + ) + ) + outcome = _TurnOutcome( + result=turn_result, + reply_node_id=turn_node.node_id, + next_active_node_id=return_node_id if turn_result.is_complete else None, ) - ) except (ChatbotInputError, ChatbotProviderError) as e: _send_failure_webhook( session_id=session_id, @@ -296,6 +486,7 @@ def process_chatbot_session_message( ) return + turn_result = outcome.result reply = turn_result.reply completion_status = turn_result.completion_status with Session(engine) as session: @@ -304,20 +495,26 @@ def process_chatbot_session_message( session_id=session_id, role=ChatbotMessageRoleEnum.ASSISTANT, content=reply, - node_id=turn_node.node_id, + node_id=outcome.reply_node_id, ) update_chatbot_session_progress( session, session_id=session_id, collected_fields=turn_result.collected_fields, completion_status=completion_status, + active_node_id=outcome.next_active_node_id, ) + logger.info( + f"[process_chatbot_session_message] Turn stored | session_id: {session_id}, " + f"reply_node_id: {outcome.reply_node_id}, " + f"next_active_node_id: {outcome.next_active_node_id}" + ) envelope = APIResponse.success_response( data={ "session_id": str(session_id), "chatbot_id": str(chatbot_id), - "node_id": turn_node.node_id, + "node_id": outcome.reply_node_id, "reply": reply, "status": session_status, "completion_status": completion_status.value, diff --git a/backend/app/services/chatbot/utils.py b/backend/app/services/chatbot/utils.py index 764f261ca..f5e5027fe 100644 --- a/backend/app/services/chatbot/utils.py +++ b/backend/app/services/chatbot/utils.py @@ -3,8 +3,11 @@ from dataclasses import dataclass, field from typing import Any, Literal +import anthropic from langchain_anthropic import ChatAnthropic +from app.services.chatbot.exceptions import ChatbotProviderError + AnthropicEffort = Literal["low", "medium", "high", "xhigh", "max"] DEFAULT_MAX_OUTPUT_TOKENS = 1024 @@ -13,6 +16,9 @@ DEFAULT_HISTORY_TOKEN_LIMIT = 4000 FLOW_KEEP_RECENT_MESSAGES = 6 +# Forced tool use is rejected with thinking on, and Opus 5.5 can't disable thinking. +STRUCTURED_OUTPUT_METHOD = "json_schema" + def default_thinking() -> dict[str, Any]: return dict(DEFAULT_THINKING) @@ -46,3 +52,25 @@ def get_chat_model(model_settings: ChatbotModelSettings) -> ChatAnthropic: if model_settings.reasoning_effort is not None: kwargs["reasoning_effort"] = model_settings.reasoning_effort return ChatAnthropic(**kwargs) + + +def to_provider_error(error: Exception, *, model: str) -> ChatbotProviderError: + if isinstance(error, anthropic.APITimeoutError): + return ChatbotProviderError( + "[KAAPI] Request timed out — retry smaller. If persistent, contact Kaapi. " + "(code: APITimeoutError)" + ) + if isinstance(error, anthropic.APIConnectionError): + return ChatbotProviderError( + "[KAAPI] Network/DNS issue reaching provider — check connectivity. " + "If persistent, contact Kaapi. (code: APIConnectionError)" + ) + if isinstance(error, anthropic.APIStatusError): + return ChatbotProviderError( + f"[ANTHROPIC] Chat model call failed: {error.message} " + f"(code: {error.status_code}, request_id: {error.request_id})" + ) + return ChatbotProviderError( + f"[KAAPI] Unsupported model configuration for '{model}': {error} " + "— update the config's model, thinking, or effort. (code: ValueError)" + ) diff --git a/backend/app/tests/api/routes/chatbot/test_flow.py b/backend/app/tests/api/routes/chatbot/test_flow.py index 06c10b9f8..aadfa98de 100644 --- a/backend/app/tests/api/routes/chatbot/test_flow.py +++ b/backend/app/tests/api/routes/chatbot/test_flow.py @@ -12,11 +12,15 @@ from app.tests.utils.auth import TestAuthContext from app.tests.utils.chatbot import ( CHATBOTS_URL, + COORDINATOR_NODE_ID, ONBOARDING_NODE_ID, UNKNOWN_CHATBOT_ID, add_test_config_version, + build_coordinator_flow, + build_coordinator_node, build_flow_edge, build_flow_node, + build_intent_edge, build_onboarding_node, create_test_anthropic_config, create_test_chatbot, @@ -34,6 +38,25 @@ MISSING_CREDENTIALS = "Anthropic credentials are not configured for this project" INPUT_NODE_ID = "input-1" OUTPUT_NODE_ID = "output-1" +ROUTES = {"Enroll": "enroll-agent", "Feedback": "feedback-agent"} + + +def _coordinator_flow_with( + cid: UUID, + *, + extra_nodes: list[dict[str, Any]] | None = None, + extra_edges: list[dict[str, Any]] | None = None, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """A valid input -> coordinator -> two-agent flow plus whatever breaks it.""" + nodes, edges = build_coordinator_flow(config_id=cid, routes=ROUTES) + return nodes + (extra_nodes or []), edges + (extra_edges or []) + + +def _coordinator_flow_replacing_node( + cid: UUID, node: dict[str, Any] +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + nodes, edges = build_coordinator_flow(config_id=cid, routes=ROUTES) + return [node if n["id"] == node["id"] else n for n in nodes], edges @pytest.fixture @@ -459,6 +482,36 @@ def test_missing_config_version_is_pinned_to_latest_and_stored( assert stored.nodes[1]["data"]["label"] == "Beneficiary Info Extractor" assert "config_version" not in stored.nodes[0]["data"] + def test_coordinator_flow_publishes_with_its_config_version_pinned( + self, + db: Session, + config: Config, + chatbot: Chatbot, + save: Callable[..., Response], + publish: Callable[[], Response], + live_v1: dict[str, Any], + ) -> None: + add_test_config_version(db, config) + nodes, edges = build_coordinator_flow(config_id=config.id, routes=ROUTES) + coordinator = next(n for n in nodes if n["id"] == COORDINATOR_NODE_ID) + del coordinator["data"]["config_version"] + assert save(nodes, edges).status_code == 200 + + resp = publish() + + assert resp.status_code == 200 + assert resp.json()["data"]["is_live"] is True + _, stored = list_stored_flow_versions(db, chatbot) + assert stored.is_live is True + stored_nodes = {node["id"]: node for node in stored.nodes} + assert stored_nodes[COORDINATOR_NODE_ID]["data"]["config_version"] == 2 + assert { + edge["data"]["intent"] for edge in stored.edges if edge.get("data") + } == { + "Enroll", + "Feedback", + } + def test_explicit_config_version_is_kept( self, db: Session, @@ -578,12 +631,93 @@ def test_unknown_chatbot_returns_404( ), "edge 'e2' source 'ghost' is not a node in the flow", ), - "coordinator-node": ( - lambda cid: ( - [build_onboarding_node(config_id=cid, node_type="coordinator")], - [], + "coordinator-without-routes": ( + lambda cid: ([build_coordinator_node(config_id=cid)], []), + f"coordinator node '{COORDINATOR_NODE_ID}' needs at least one outgoing edge", + ), + "coordinator-route-to-rag": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[ + build_onboarding_node(config_id=cid, node_id="faq", node_type="rag") + ], + extra_edges=[build_intent_edge("e-faq", "faq", "Questions")], + ), + f"edge 'e-faq' from coordinator '{COORDINATOR_NODE_ID}' must target an " + "onboarding node (target 'faq' is 'rag')", + ), + "coordinator-route-to-output": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_flow_node(OUTPUT_NODE_ID, "output")], + extra_edges=[build_intent_edge("e-out", OUTPUT_NODE_ID, "Done")], + ), + f"edge 'e-out' from coordinator '{COORDINATOR_NODE_ID}' must target an " + f"onboarding node (target '{OUTPUT_NODE_ID}' is 'output')", + ), + "coordinator-route-to-coordinator": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_coordinator_node(config_id=cid, node_id="router-2")], + extra_edges=[ + build_intent_edge("e-router-2", "router-2", "Other"), + build_intent_edge( + "e-back", "enroll-agent", "Enroll", source="router-2" + ), + ], + ), + f"edge 'e-router-2' from coordinator '{COORDINATOR_NODE_ID}' must target an " + "onboarding node (target 'router-2' is 'coordinator')", + ), + "two-coordinators": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_coordinator_node(config_id=cid, node_id="router-2")], + extra_edges=[ + build_intent_edge( + "e-router-2", "feedback-agent", "Feedback", source="router-2" + ) + ], ), - f"node '{ONBOARDING_NODE_ID}' has type 'coordinator', which is not supported yet", + "at most 1 coordinator node allowed (found 2)", + ), + "coordinator-route-missing-intent": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_onboarding_node(config_id=cid, node_id="third")], + extra_edges=[build_flow_edge("e-third", COORDINATOR_NODE_ID, "third")], + ), + f"edge 'e-third' from coordinator '{COORDINATOR_NODE_ID}' is missing " + "data.intent", + ), + "coordinator-route-blank-intent": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_onboarding_node(config_id=cid, node_id="third")], + extra_edges=[build_intent_edge("e-third", "third", " ")], + ), + f"edge 'e-third' from coordinator '{COORDINATOR_NODE_ID}' is missing " + "data.intent", + ), + "coordinator-duplicate-intent": ( + lambda cid: _coordinator_flow_with( + cid, + extra_nodes=[build_onboarding_node(config_id=cid, node_id="third")], + extra_edges=[build_intent_edge("e-third", "third", " Enroll ")], + ), + f"coordinator '{COORDINATOR_NODE_ID}' has duplicate intent 'Enroll'", + ), + "coordinator-missing-prompt": ( + lambda cid: _coordinator_flow_replacing_node( + cid, build_coordinator_node(config_id=cid, prompt=None) + ), + f"node '{COORDINATOR_NODE_ID}' is missing data.prompt", + ), + "coordinator-missing-config-id": ( + lambda cid: _coordinator_flow_replacing_node( + cid, build_coordinator_node(config_id=None) + ), + f"node '{COORDINATOR_NODE_ID}' is missing data.config_id", ), "rag-node": ( lambda cid: ([build_onboarding_node(config_id=cid, node_type="rag")], []), @@ -663,7 +797,7 @@ def test_unknown_chatbot_returns_404( # A representative subset: the first save shares publish's validator, so the full # matrix is covered by the publish tests below. FIRST_SAVE_INVALID_CASES = [ - "coordinator-node", + "coordinator-route-missing-intent", "duplicate-node-ids", "input-edge-to-output", "no-edges-no-agent", @@ -778,9 +912,7 @@ def test_all_problems_are_reported_together( [ build_flow_node("in-a", "input"), build_flow_node("in-b", "input"), - build_onboarding_node( - config_id=config.id, node_id="router", node_type="coordinator" - ), + build_coordinator_node(config_id=config.id, node_id="router"), build_onboarding_node(config_id=config.id, prompt=None), ] ) @@ -790,7 +922,7 @@ def test_all_problems_are_reported_together( assert resp.status_code == 422 error = resp.json()["error"] assert "at most 1 input node allowed (found 2)" in error - assert "node 'router' has type 'coordinator'" in error + assert "coordinator node 'router' needs at least one outgoing edge" in error assert f"node '{ONBOARDING_NODE_ID}' is missing data.prompt" in error assert error.count(INVALID_FLOW_PREFIX) == 1 diff --git a/backend/app/tests/crud/test_chatbot.py b/backend/app/tests/crud/test_chatbot.py index fef300e2b..631267f7e 100644 --- a/backend/app/tests/crud/test_chatbot.py +++ b/backend/app/tests/crud/test_chatbot.py @@ -43,6 +43,7 @@ ChatbotSessionConflictError, ) from app.tests.utils.chatbot import ( + COORDINATOR_NODE_ID, ONBOARDING_NODE_ID, add_test_chatbot_messages, build_onboarding_node, @@ -564,6 +565,28 @@ def test_replaces_rather_than_merges_fields( assert stored.collected_fields == {"name": "Asha"} assert stored.completion_status == ChatbotCompletionStatusEnum.IN_PROGRESS + def test_active_node_moves_only_when_given( + self, db: Session, chatbot: Chatbot + ) -> None: + chatbot_session = create_test_chatbot_session(db, chatbot=chatbot) + + update_chatbot_session_progress( + db, + session_id=chatbot_session.id, + collected_fields={}, + completion_status=ChatbotCompletionStatusEnum.IN_PROGRESS, + active_node_id=COORDINATOR_NODE_ID, + ) + assert _reload(db, chatbot_session).active_node_id == COORDINATOR_NODE_ID + + update_chatbot_session_progress( + db, + session_id=chatbot_session.id, + collected_fields={}, + completion_status=ChatbotCompletionStatusEnum.IN_PROGRESS, + ) + assert _reload(db, chatbot_session).active_node_id == COORDINATOR_NODE_ID + def test_unknown_id_returns_none(self, db: Session) -> None: assert ( update_chatbot_session_progress( diff --git a/backend/app/tests/services/chatbot/agents/coordinator/__init__.py b/backend/app/tests/services/chatbot/agents/coordinator/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/tests/services/chatbot/agents/coordinator/test_graph.py b/backend/app/tests/services/chatbot/agents/coordinator/test_graph.py new file mode 100644 index 000000000..c923e3160 --- /dev/null +++ b/backend/app/tests/services/chatbot/agents/coordinator/test_graph.py @@ -0,0 +1,216 @@ +import anthropic +import httpx +import pytest +from langchain_core.exceptions import OutputParserException + +from app.services.chatbot.agents.coordinator.graph import ( + CoordinatorDecision, + CoordinatorRoute, + RouteDecision, + run_coordinator_turn, +) +from app.services.chatbot.agents.coordinator.prompts import ( + FALLBACK_CLARIFYING_QUESTION, +) +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.services.chatbot.types import ChatbotMessageDict +from app.services.chatbot.utils import ChatbotModelSettings +from app.tests.utils.chatbot import ( + ANTHROPIC_REQUEST, + COORDINATOR_PROMPT, + ScriptedChatModel, + patch_chat_model, +) + +MODEL = "claude-sonnet-4-6" +ROUTES = [ + CoordinatorRoute( + intent="Enroll", description="Sign up a new student", target_node_id="enroll" + ), + CoordinatorRoute(intent="Feedback", description=None, target_node_id="feedback"), +] +CLARIFY = CoordinatorDecision( + target_node_id=None, clarifying_question="Do you want to enroll or give feedback?" +) + + +async def _run( + *, + user_message: str | None = "I want to enroll my son", + history: list[ChatbotMessageDict] | None = None, +) -> CoordinatorDecision: + return await run_coordinator_turn( + system_prompt=COORDINATOR_PROMPT, + history=history or [], + user_message=user_message, + routes=ROUTES, + model_settings=ChatbotModelSettings(model=MODEL), + ) + + +def _scripted(monkeypatch: pytest.MonkeyPatch, *outcomes: object) -> ScriptedChatModel: + return patch_chat_model( + monkeypatch, + ScriptedChatModel(replies=["should never be used"], extractions=list(outcomes)), + ) + + +class TestRouting: + @pytest.mark.parametrize( + ("intent", "target"), + [ + ("Enroll", "enroll"), + ("Feedback", "feedback"), + (" enroll ", "enroll"), + ("FEEDBACK", "feedback"), + ], + ids=["exact", "second-route", "padded-lowercase", "uppercase"], + ) + @pytest.mark.asyncio + async def test_matching_intent_routes_to_its_target( + self, monkeypatch: pytest.MonkeyPatch, intent: str, target: str + ) -> None: + _scripted(monkeypatch, RouteDecision(intent=intent)) + + decision = await _run() + + assert decision == CoordinatorDecision( + target_node_id=target, clarifying_question=None + ) + + @pytest.mark.parametrize( + "intent", [None, "Billing"], ids=["null-intent", "unknown-intent"] + ) + @pytest.mark.asyncio + async def test_no_matching_intent_asks_the_models_question( + self, monkeypatch: pytest.MonkeyPatch, intent: str | None + ) -> None: + _scripted( + monkeypatch, + RouteDecision( + intent=intent, clarifying_question=CLARIFY.clarifying_question + ), + ) + + assert await _run(user_message="hmm") == CLARIFY + + @pytest.mark.parametrize("question", [None, " "], ids=["missing", "blank"]) + @pytest.mark.asyncio + async def test_missing_question_falls_back_to_default( + self, monkeypatch: pytest.MonkeyPatch, question: str | None + ) -> None: + _scripted(monkeypatch, RouteDecision(intent=None, clarifying_question=question)) + + decision = await _run(user_message="hmm") + + assert decision == CoordinatorDecision( + target_node_id=None, clarifying_question=FALLBACK_CLARIFYING_QUESTION + ) + + @pytest.mark.asyncio + async def test_opening_turn_clarifies_even_when_model_names_an_intent( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # The opening turn only carries the synthetic kickoff, so there's nothing to route on. + _scripted( + monkeypatch, + RouteDecision(intent="Enroll", clarifying_question="Hi! How can I help?"), + ) + + decision = await _run(user_message=None) + + assert decision == CoordinatorDecision( + target_node_id=None, clarifying_question="Hi! How can I help?" + ) + + @pytest.mark.asyncio + async def test_routing_is_one_json_schema_call_listing_every_intent( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = _scripted(monkeypatch, RouteDecision(intent="Enroll")) + + await _run() + + assert chat_model.structured_output_schemas == [RouteDecision] + # Forced tool use is rejected while thinking is on. + assert chat_model.structured_output_methods == ["json_schema"] + assert chat_model.calls == [] + [sent] = chat_model.extraction_calls + system_prompt = str(sent[0].content) + assert system_prompt.startswith(COORDINATOR_PROMPT) + assert "- Enroll: Sign up a new student" in system_prompt.splitlines() + assert "- Feedback" in system_prompt.splitlines() + assert sent[-1].content == "I want to enroll my son" + + @pytest.mark.asyncio + async def test_missing_message_with_history_rejected( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = _scripted(monkeypatch) + + with pytest.raises(ChatbotInputError, match="`message` is required"): + await _run( + user_message=None, history=[{"role": "assistant", "content": "Hi!"}] + ) + assert chat_model.extraction_calls == [] + + +class TestRoutingErrors: + @pytest.mark.parametrize( + ("outcome", "expected_message"), + [ + ( + OutputParserException("not valid JSON"), + "[ANTHROPIC] Coordinator routing returned malformed structured output", + ), + ( + {"intent": "Enroll"}, + "[ANTHROPIC] Coordinator routing returned an unexpected output shape", + ), + ( + ValueError("effort is not supported for this model"), + f"[KAAPI] Unsupported model configuration for '{MODEL}': " + "effort is not supported for this model", + ), + ( + anthropic.APITimeoutError(request=ANTHROPIC_REQUEST), + "(code: APITimeoutError)", + ), + ( + anthropic.APIConnectionError(request=ANTHROPIC_REQUEST), + "(code: APIConnectionError)", + ), + ( + anthropic.APIStatusError( + "Overloaded", + response=httpx.Response( + 529, request=ANTHROPIC_REQUEST, headers={"request-id": "req_9"} + ), + body=None, + ), + "[ANTHROPIC] Chat model call failed: Overloaded " + "(code: 529, request_id: req_9)", + ), + ], + ids=[ + "output_parser", + "wrong_type", + "value_error", + "timeout", + "connection", + "status", + ], + ) + @pytest.mark.asyncio + async def test_routing_failure_becomes_provider_error( + self, + monkeypatch: pytest.MonkeyPatch, + outcome: object, + expected_message: str, + ) -> None: + _scripted(monkeypatch, outcome) + + with pytest.raises(ChatbotProviderError) as exc_info: + await _run() + + assert expected_message in str(exc_info.value) diff --git a/backend/app/tests/services/chatbot/test_flow.py b/backend/app/tests/services/chatbot/test_flow.py index 1bce0facb..adfd06a0b 100644 --- a/backend/app/tests/services/chatbot/test_flow.py +++ b/backend/app/tests/services/chatbot/test_flow.py @@ -10,6 +10,7 @@ from app.tests.utils.chatbot import ( ONBOARDING_NODE_ID, add_test_config_version, + build_coordinator_flow, build_flow_edge, build_flow_node, build_onboarding_node, @@ -283,3 +284,19 @@ def test_missing_anthropic_credentials_joins_the_other_problems( assert "at most 1 output node allowed (found 2)" in message assert "Anthropic credentials are not configured for this project" in message assert message.count("[KAAPI]") == 1 + + def test_intents_differing_only_by_case_are_duplicates( + self, db: Session, config: Config + ) -> None: + nodes, edges = build_coordinator_flow( + config_id=config.id, routes={"Enroll": "first", "enroll": "second"} + ) + + with pytest.raises(ChatbotFlowValidationError, match="duplicate intent"): + validate_flow_for_publish( + session=db, + organization_id=_organization_id(db, config), + project_id=config.project_id, + nodes=[ChatbotFlowNode.model_validate(n) for n in nodes], + edges=[ChatbotFlowEdge.model_validate(e) for e in edges], + ) diff --git a/backend/app/tests/services/chatbot/test_session.py b/backend/app/tests/services/chatbot/test_session.py index 295a8a8fc..e92529f7f 100644 --- a/backend/app/tests/services/chatbot/test_session.py +++ b/backend/app/tests/services/chatbot/test_session.py @@ -5,6 +5,7 @@ import anthropic import pytest +from langchain_core.exceptions import OutputParserException from sqlmodel import Session from app.crud.chatbot import ( @@ -21,6 +22,10 @@ Project, ) from app.services.chatbot import session as session_service +from app.services.chatbot.agents.coordinator.graph import RouteDecision +from app.services.chatbot.agents.coordinator.prompts import ( + FALLBACK_CLARIFYING_QUESTION, +) from app.services.chatbot.agents.onboarding import graph as onboarding_graph from app.services.chatbot.agents.onboarding.types import FieldExtraction from app.services.chatbot.session import ( @@ -30,6 +35,7 @@ from app.services.chatbot.utils import ChatbotModelSettings from app.tests.utils.chatbot import ( ANTHROPIC_REQUEST, + COORDINATOR_NODE_ID, DEFAULT_SYSTEM_PROMPT, ONBOARDING_NODE_ID, WEBHOOK_URL, @@ -42,8 +48,10 @@ create_test_chatbot_session, create_test_flow_version, create_test_live_chatbot, + create_test_live_coordinator_chatbot, extracted_fields, patch_chat_model, + route_prompt, ) from app.tests.utils.test_data import create_test_config @@ -140,13 +148,14 @@ def _success_webhook( reply: str, completion_status: str = "in_progress", collected_fields: dict[str, str] | None = None, + node_id: str = ONBOARDING_NODE_ID, ) -> dict[str, Any]: return { "success": True, "data": { "session_id": str(chatbot_session.id), "chatbot_id": str(chatbot.id), - "node_id": ONBOARDING_NODE_ID, + "node_id": node_id, "reply": reply, "status": "active", "completion_status": completion_status, @@ -705,27 +714,27 @@ def test_active_node_missing_from_flow_posts_failure_webhook( ) ] - def test_non_onboarding_active_node_posts_failure_webhook( + def test_rag_active_node_posts_failure_webhook( self, db: Session, config: Config, callbacks: CallbackRecorder, unused_chat_model: ScriptedChatModel, ) -> None: - coordinator = build_onboarding_node( - config_id=config.id, node_id="router", node_type="coordinator" - ) + rag = build_onboarding_node(config_id=config.id, node_id="faq", node_type="rag") chatbot_session = _session_on_live_flow( - db, config, nodes=[coordinator], active_node_id="router" + db, config, nodes=[rag], active_node_id="faq" ) _process(chatbot_session, db) assert unused_chat_model.calls == [] - [payload] = callbacks.payloads - assert payload["success"] is False - assert "'router' has type 'coordinator'" in payload["error"] - assert "only 'onboarding' nodes can run" in payload["error"] + _assert_single_failure_webhook( + callbacks, + chatbot_session, + "Active node 'faq' has type 'rag'", + "only 'coordinator', 'onboarding' nodes can run", + ) @pytest.mark.parametrize( "node_overrides", @@ -754,3 +763,333 @@ def test_node_without_prompt_or_config_posts_failure_webhook( _assert_single_failure_webhook( callbacks, chatbot_session, "missing data.prompt or data.config_id" ) + + +ENROLL_NODE_ID = "enroll-agent" +FEEDBACK_NODE_ID = "feedback-agent" +COORDINATOR_ROUTES = {"Enroll": ENROLL_NODE_ID, "Feedback": FEEDBACK_NODE_ID} +GREETING = "Hi! Do you want to enroll or share feedback?" + + +@pytest.fixture +def coordinator_chatbot(db: Session, config: Config) -> Chatbot: + return create_test_live_coordinator_chatbot( + db, config=config, routes=COORDINATOR_ROUTES + )[0] + + +def _coordinator_session( + db: Session, + chatbot: Chatbot, + turns: ChatTurns, + *, + active_node_id: str = COORDINATOR_NODE_ID, +) -> ChatbotSession: + chatbot_session = create_test_chatbot_session( + db, chatbot=chatbot, active_node_id=active_node_id + ) + add_test_chatbot_messages(db, chatbot_session, turns) + return chatbot_session + + +def _reload(db: Session, chatbot_session: ChatbotSession) -> ChatbotSession: + db.expire_all() + stored = db.get(ChatbotSession, chatbot_session.id) + assert stored is not None + return stored + + +def _assistant_node_ids( + db: Session, chatbot_session: ChatbotSession +) -> list[str | None]: + return [ + m.node_id + for m in list_chatbot_session_messages(db, session_id=chatbot_session.id) + if m.role == ChatbotMessageRoleEnum.ASSISTANT + ] + + +class TestCoordinatorTurns: + def test_opening_turn_clarifies_and_stays_on_coordinator( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + # An intent on the opening turn must not hand off: there's no user message yet. + chat_model = patch_chat_model( + monkeypatch, + ScriptedChatModel( + extractions=[ + RouteDecision(intent="Enroll", clarifying_question=GREETING) + ] + ), + ) + chatbot_session = _coordinator_session(db, coordinator_chatbot, []) + + _process(chatbot_session, db) + + assert chat_model.structured_output_schemas == [RouteDecision] + assert chat_model.calls == [] + assert _transcript(db, chatbot_session) == [("assistant", GREETING)] + assert _assistant_node_ids(db, chatbot_session) == [COORDINATOR_NODE_ID] + assert callbacks.payloads == [ + _success_webhook( + chatbot_session, + coordinator_chatbot, + reply=GREETING, + node_id=COORDINATOR_NODE_ID, + ) + ] + stored = _reload(db, chatbot_session) + assert stored.active_node_id == COORDINATOR_NODE_ID + assert stored.completion_status == ChatbotCompletionStatusEnum.IN_PROGRESS + + def test_ambiguous_message_asks_fallback_question_and_keeps_fields( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + patch_chat_model( + monkeypatch, + ScriptedChatModel(extractions=[RouteDecision(intent=None)]), + ) + chatbot_session = _coordinator_session( + db, + coordinator_chatbot, + [ + (ChatbotMessageRoleEnum.ASSISTANT, GREETING), + (ChatbotMessageRoleEnum.USER, "hmm"), + ], + ) + update_chatbot_session_progress( + db, + session_id=chatbot_session.id, + collected_fields={"name": "Asha"}, + completion_status=ChatbotCompletionStatusEnum.IN_PROGRESS, + ) + + _process(chatbot_session, db) + + assert callbacks.payloads == [ + _success_webhook( + chatbot_session, + coordinator_chatbot, + reply=FALLBACK_CLARIFYING_QUESTION, + node_id=COORDINATOR_NODE_ID, + collected_fields={"name": "Asha"}, + ) + ] + assert _assistant_node_ids(db, chatbot_session) == [ + COORDINATOR_NODE_ID, + COORDINATOR_NODE_ID, + ] + stored = _reload(db, chatbot_session) + assert stored.active_node_id == COORDINATOR_NODE_ID + assert stored.collected_fields == {"name": "Asha"} + + def test_matched_intent_hands_off_to_target_in_the_same_turn( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + chat_model = patch_chat_model( + monkeypatch, + ScriptedChatModel( + replies=["Great, what's your child's name?"], + extractions=[ + RouteDecision(intent="enroll"), + FieldExtraction( + collected_fields=extracted_fields({"parent": "Asha"}), + missing_fields=["child"], + is_complete=False, + ), + ], + ), + ) + chatbot_session = _coordinator_session( + db, + coordinator_chatbot, + [ + (ChatbotMessageRoleEnum.ASSISTANT, GREETING), + (ChatbotMessageRoleEnum.USER, "I'm Asha, I want to enroll my son"), + ], + ) + + _process(chatbot_session, db) + + assert chat_model.structured_output_schemas == [RouteDecision, FieldExtraction] + [reply_call] = chat_model.calls + assert str(reply_call[0].content).startswith(route_prompt(ENROLL_NODE_ID)) + assert _assistant_node_ids(db, chatbot_session) == [ + COORDINATOR_NODE_ID, + ENROLL_NODE_ID, + ] + assert callbacks.payloads == [ + _success_webhook( + chatbot_session, + coordinator_chatbot, + reply="Great, what's your child's name?", + node_id=ENROLL_NODE_ID, + collected_fields={"parent": "Asha"}, + ) + ] + stored = _reload(db, chatbot_session) + assert stored.active_node_id == ENROLL_NODE_ID + assert stored.completion_status == ChatbotCompletionStatusEnum.IN_PROGRESS + + def test_target_completing_in_the_routed_turn_returns_to_coordinator( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + patch_chat_model( + monkeypatch, + ScriptedChatModel( + replies=["Thanks for the feedback!"], + extractions=[ + RouteDecision(intent="Feedback"), + FieldExtraction( + collected_fields=extracted_fields({"rating": "5"}), + missing_fields=[], + is_complete=True, + ), + ], + ), + ) + chatbot_session = _coordinator_session( + db, + coordinator_chatbot, + [ + (ChatbotMessageRoleEnum.ASSISTANT, GREETING), + (ChatbotMessageRoleEnum.USER, "Feedback: 5 stars"), + ], + ) + + _process(chatbot_session, db) + + [payload] = callbacks.payloads + assert payload["data"]["node_id"] == FEEDBACK_NODE_ID + assert payload["data"]["completion_status"] == "complete" + stored = _reload(db, chatbot_session) + assert stored.active_node_id == COORDINATOR_NODE_ID + assert stored.completion_status == ChatbotCompletionStatusEnum.COMPLETE + assert stored.collected_fields == {"rating": "5"} + + def test_agent_completing_on_a_later_turn_returns_to_coordinator( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + chat_model = patch_chat_model( + monkeypatch, + ScriptedChatModel( + replies=["You're enrolled!"], + extractions=[ + FieldExtraction( + collected_fields=extracted_fields({"child": "Ravi"}), + missing_fields=[], + is_complete=True, + ) + ], + ), + ) + chatbot_session = _coordinator_session( + db, + coordinator_chatbot, + [ + (ChatbotMessageRoleEnum.ASSISTANT, "What's your child's name?"), + (ChatbotMessageRoleEnum.USER, "Ravi"), + ], + active_node_id=ENROLL_NODE_ID, + ) + + _process(chatbot_session, db) + + assert chat_model.structured_output_schemas == [FieldExtraction] + [payload] = callbacks.payloads + assert payload["data"]["node_id"] == ENROLL_NODE_ID + assert payload["data"]["completion_status"] == "complete" + stored = _reload(db, chatbot_session) + assert stored.active_node_id == COORDINATOR_NODE_ID + assert stored.completion_status == ChatbotCompletionStatusEnum.COMPLETE + + def test_entry_onboarding_flow_stays_on_its_node_after_completion( + self, + db: Session, + chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + patch_chat_model( + monkeypatch, + ScriptedChatModel( + replies=["All set!"], + extractions=[ + FieldExtraction( + collected_fields=extracted_fields({"name": "Asha"}), + missing_fields=[], + is_complete=True, + ) + ], + ), + ) + chatbot_session = _session_with_turns( + db, chatbot, [(ChatbotMessageRoleEnum.USER, "Asha")] + ) + + _process(chatbot_session, db) + + stored = _reload(db, chatbot_session) + assert stored.active_node_id == ONBOARDING_NODE_ID + assert stored.completion_status == ChatbotCompletionStatusEnum.COMPLETE + assert callbacks.payloads[0]["data"]["node_id"] == ONBOARDING_NODE_ID + + @pytest.mark.parametrize( + ("routing_error", "expected_message"), + [ + ( + anthropic.APITimeoutError(request=ANTHROPIC_REQUEST), + "(code: APITimeoutError)", + ), + ( + OutputParserException("not valid JSON"), + "[ANTHROPIC] Coordinator routing returned malformed structured output", + ), + ], + ids=["timeout", "malformed"], + ) + def test_routing_failure_posts_failure_webhook_and_stores_nothing( + self, + db: Session, + coordinator_chatbot: Chatbot, + callbacks: CallbackRecorder, + monkeypatch: pytest.MonkeyPatch, + routing_error: Exception, + expected_message: str, + ) -> None: + chat_model = patch_chat_model( + monkeypatch, + ScriptedChatModel( + replies=["should never be used"], extractions=[routing_error] + ), + ) + chatbot_session = _coordinator_session( + db, coordinator_chatbot, [(ChatbotMessageRoleEnum.USER, "enroll")] + ) + + _process(chatbot_session, db) + + assert chat_model.calls == [] + _assert_single_failure_webhook(callbacks, chatbot_session, expected_message) + assert _transcript(db, chatbot_session) == [("user", "enroll")] + assert _reload(db, chatbot_session).active_node_id == COORDINATOR_NODE_ID diff --git a/backend/app/tests/utils/chatbot.py b/backend/app/tests/utils/chatbot.py index 7a9b68ba2..936f05a8a 100644 --- a/backend/app/tests/utils/chatbot.py +++ b/backend/app/tests/utils/chatbot.py @@ -35,6 +35,7 @@ Project, ) from app.models.provider_credentials import Provider +from app.services.chatbot.agents.coordinator import graph as coordinator_graph from app.services.chatbot.agents.onboarding import graph as onboarding_graph from app.services.chatbot.agents.onboarding.types import ( ExtractedField, @@ -51,6 +52,9 @@ DEFAULT_SYSTEM_PROMPT = "Collect the user's name and city." ONBOARDING_NODE_ID = "onboarding-1" ONBOARDING_NODE_LABEL = "Beneficiary Info Extractor" +INPUT_NODE_ID = "input-1" +COORDINATOR_NODE_ID = "coordinator-1" +COORDINATOR_PROMPT = "You greet users and route them to the right assistant." # A model other than build_anthropic_text_config_blob's default, so a newer # config version is distinguishable from version 1. NEWER_CONFIG_MODEL = "claude-opus-5-5" @@ -107,6 +111,8 @@ class ScriptedChatModel(BaseChatModel): extractions: list[Any] = Field(default_factory=list) extraction_calls: list[list[BaseMessage]] = Field(default_factory=list) structured_output_methods: list[str | None] = Field(default_factory=list) + # Schema of each structured call actually invoked, in call order. + structured_output_schemas: list[Any] = Field(default_factory=list) @property def _llm_type(self) -> str: @@ -137,6 +143,7 @@ def with_structured_output( def _extract(messages: list[BaseMessage]) -> Any: self.extraction_calls.append(list(messages)) + self.structured_output_schemas.append(schema) outcome = self.extractions.pop(0) if self.extractions else NOTHING_EXTRACTED if isinstance(outcome, BaseException): raise outcome @@ -149,9 +156,8 @@ def patch_chat_model( monkeypatch: pytest.MonkeyPatch, chat_model: ScriptedChatModel ) -> ScriptedChatModel: # The Anthropic SDK is the external boundary; everything above it runs for real. - monkeypatch.setattr( - onboarding_graph, "get_chat_model", lambda _settings: chat_model - ) + for agent_graph in (onboarding_graph, coordinator_graph): + monkeypatch.setattr(agent_graph, "get_chat_model", lambda _settings: chat_model) return chat_model @@ -223,8 +229,19 @@ def build_flow_node(node_id: str, node_type: str, **data: Any) -> dict[str, Any] } -def build_flow_edge(edge_id: str, source: str, target: str) -> dict[str, Any]: - return {"id": edge_id, "source": source, "target": target} +def build_flow_edge( + edge_id: str, source: str, target: str, *, data: dict[str, Any] | None = None +) -> dict[str, Any]: + edge: dict[str, Any] = {"id": edge_id, "source": source, "target": target} + if data is not None: + edge["data"] = data + return edge + + +def build_intent_edge( + edge_id: str, target: str, intent: str, *, source: str = COORDINATOR_NODE_ID +) -> dict[str, Any]: + return build_flow_edge(edge_id, source, target, data={"intent": intent}) def build_onboarding_node( @@ -245,6 +262,45 @@ def build_onboarding_node( return build_flow_node(node_id, node_type, **data) +def build_coordinator_node( + *, + config_id: UUID | None, + config_version: int | None = 1, + prompt: str | None = COORDINATOR_PROMPT, + node_id: str = COORDINATOR_NODE_ID, +) -> dict[str, Any]: + return build_onboarding_node( + config_id=config_id, + config_version=config_version, + prompt=prompt, + node_id=node_id, + node_type=ChatbotFlowNodeTypeEnum.COORDINATOR.value, + ) + + +def build_coordinator_flow( + *, config_id: UUID, routes: dict[str, str] +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """input -> coordinator -> one onboarding node per `routes` entry (intent -> node id).""" + nodes = [ + build_flow_node(INPUT_NODE_ID, ChatbotFlowNodeTypeEnum.INPUT.value), + build_coordinator_node(config_id=config_id), + ] + edges = [build_flow_edge("e-input", INPUT_NODE_ID, COORDINATOR_NODE_ID)] + for intent, node_id in routes.items(): + nodes.append( + build_onboarding_node( + config_id=config_id, node_id=node_id, prompt=route_prompt(node_id) + ) + ) + edges.append(build_intent_edge(f"e-{node_id}", node_id, intent)) + return nodes, edges + + +def route_prompt(node_id: str) -> str: + return f"Collect the details needed by {node_id}." + + def create_test_chatbot( db: Session, *, @@ -309,6 +365,20 @@ def create_test_live_chatbot( return chatbot, flow_version +def create_test_live_coordinator_chatbot( + db: Session, *, config: Config, routes: dict[str, str] +) -> tuple[Chatbot, ChatbotFlowVersion]: + """A chatbot whose live flow is input -> coordinator -> one onboarding node per route.""" + project = db.get(Project, config.project_id) + assert project is not None + chatbot = create_test_chatbot(db, project=project) + nodes, edges = build_coordinator_flow(config_id=config.id, routes=routes) + flow_version = create_test_flow_version( + db, chatbot=chatbot, nodes=nodes, edges=edges, is_live=True + ) + return chatbot, flow_version + + def create_test_chatbot_session( db: Session, *, diff --git a/docs/wiki/domain-map.md b/docs/wiki/domain-map.md index 37759e939..19acdd8dd 100644 --- a/docs/wiki/domain-map.md +++ b/docs/wiki/domain-map.md @@ -53,7 +53,7 @@ APIKey → Organization, Project, User # programmatic access | Chatbot | chatbot.py | Org, Project | ChatbotFlowVersion, ChatbotSession; soft-deleted (`deleted_at`) because sessions reference it | | ChatbotFlowVersion | chatbot.py | Chatbot (CASCADE) | ChatbotSession (`flow_version_id`, RESTRICT); React Flow editor in kaapi-frontend (logical); Config/ConfigVersion via node `data` (logical) | | ChatbotSession | chatbot.py | Chatbot (RESTRICT), ChatbotFlowVersion (RESTRICT) | ChatbotSessionMessage; Celery turn worker; caller webhook (logical) | -| ChatbotSessionMessage | chatbot.py | ChatbotSession (CASCADE) | turn history for the onboarding agent (logical) | +| ChatbotSessionMessage | chatbot.py | ChatbotSession (CASCADE) | turn history for the coordinator + onboarding agents (logical) | ## External consumers (always check in blast radius) diff --git a/docs/wiki/modules/chatbot.md b/docs/wiki/modules/chatbot.md index efaeb4df7..ac08cc0e6 100644 --- a/docs/wiki/modules/chatbot.md +++ b/docs/wiki/modules/chatbot.md @@ -1,6 +1,6 @@ # Module: Chatbot -A project has many chatbots; each chatbot's behaviour is a React Flow graph (nodes + edges JSON) edited in the UI and versioned with draft/publish, so editing never breaks running conversations. Agent nodes carry their own `prompt` and saved Kaapi config (`config_id` + `config_version`, pinned at publish); only Anthropic text completion configs are supported. Conversations are server-side sessions pinned to the live flow version and an active node; turns run on Celery and reply via webhook. On this branch the runtime only executes `onboarding` agent nodes ("Beneficiary Info Extractor" over a two-node LangGraph): the node prompt defines the goal, fields to collect and rules; each turn extracts the fields given so far and then replies. No deep-dive doc yet. +A project has many chatbots; each chatbot's behaviour is a React Flow graph (nodes + edges JSON) edited in the UI and versioned with draft/publish, so editing never breaks running conversations. Agent nodes carry their own `prompt` and saved Kaapi config (`config_id` + `config_version`, pinned at publish); only Anthropic text completion configs are supported. Conversations are server-side sessions pinned to the live flow version and an active node; turns run on Celery and reply via webhook. The runtime executes `onboarding` agent nodes ("Beneficiary Info Extractor" over a two-node LangGraph: the node prompt defines the goal, fields to collect and rules; each turn extracts the fields given so far and then replies) and a `coordinator` node that routes each message to an onboarding node by edge intent. `rag` nodes are stored but not runnable. No deep-dive doc yet. All paths relative to `backend/app/`. @@ -28,8 +28,8 @@ All paths relative to `backend/app/`. - Session lifecycle: one active session per contact + chatbot. `active`+`in_progress` → `active`+`complete` as fields are collected. Creating a new session for the same chatbot + `contact_ref` (same transaction as the insert) — or deleting the chatbot — moves active ones to `inactive`+`complete`, or `inactive`+`abandoned` if still in progress. Inactive sessions reject new messages (409); a turn already queued is skipped by the worker, which posts a failure webhook instead. A concurrent double-create hits the partial unique index; crud raises `ChatbotSessionConflictError` → 409. ## Flow JSON (Pydantic in `models/chatbot.py`) -- `ChatbotFlowNode` {id (1..255), type `ChatbotFlowNodeTypeEnum` (input/output/coordinator/onboarding/rag), position {x, y}, data}; agent nodes (`AGENT_NODE_TYPES`) validate `data` as `ChatbotAgentNodeData` {config_id: UUID?, config_version: int ≥ 1?, prompt?}. `ChatbotFlowEdge` {id, source, target, sourceHandle?, targetHandle?, data} — `data.intent` / `data.description` are reserved for coordinator routing. Extra keys allowed everywhere; dumped by alias. -- The first save goes live directly (validated + pinned like a publish); later draft saves validate shape only. Publish (`services/chatbot/flow.py:validate_flow_for_publish`) is strict: unique node ids; ≤ 1 input and ≤ 1 output node; edge endpoints exist; coordinator/rag rejected as not supported yet; every onboarding node has `config_id` + non-empty `prompt` and its config passes `get_chatbot_config_version` + `build_chatbot_model_settings`; a null `config_version` is resolved to the latest and written back into the stored node; an entry node must resolve. +- `ChatbotFlowNode` {id (1..255), type `ChatbotFlowNodeTypeEnum` (input/output/coordinator/onboarding/rag), position {x, y}, data}; agent nodes (`AGENT_NODE_TYPES`) validate `data` as `ChatbotAgentNodeData` {config_id: UUID?, config_version: int ≥ 1?, prompt?}. `ChatbotFlowEdge` {id, source, target, sourceHandle?, targetHandle?, data} — `data.intent` (required on coordinator edges at publish) / `data.description` (optional) drive coordinator routing. Extra keys allowed everywhere; dumped by alias. +- The first save goes live directly (validated + pinned like a publish); later draft saves validate shape only. Publish (`services/chatbot/flow.py:validate_flow_for_publish`) is strict: unique node ids; ≤ 1 input and ≤ 1 output node; edge endpoints exist; rag rejected as not supported yet; `_list_coordinator_problems`: ≤ 1 coordinator, ≥ 1 outgoing edge, every coordinator edge targets an onboarding node, non-empty unique `data.intent` per coordinator; every onboarding/coordinator node has `config_id` + non-empty `prompt` and its config passes `get_chatbot_config_version` + `build_chatbot_model_settings`; a null `config_version` is resolved to the latest and written back into the stored node; an entry node must resolve. - Entry node (`resolve_entry_node`): no edges → the single agent node; otherwise the agent node targeted by the input node's only outgoing edge. ## Turn flow (onboarding agent, `services/chatbot/agents/onboarding/graph.py`) @@ -37,22 +37,28 @@ All paths relative to `backend/app/`. 2. `chatbot_node` — replies using the same summarized view plus a system note listing collected/missing fields, so it asks only for what's missing and wraps up when complete. 3. `run_chatbot_turn(..., collected_fields=prior)` → `ChatbotTurnResult{reply, collected_fields, is_complete}` (`reply` = text of the graph's final `AIMessage` via `history.extract_final_reply`; empty → `ChatbotProviderError` `EmptyReply`) (`.completion_status` maps to the enum). Anthropic errors from either call, and langchain_anthropic's client-side model-config `ValueError`s ("[KAAPI] Unsupported model configuration"), map to `ChatbotProviderError`, so the session path always sends a failure webhook. +## Turn flow (coordinator, `services/chatbot/agents/coordinator/graph.py`) +- `run_coordinator_turn(..., routes)` → `CoordinatorDecision{target_node_id | clarifying_question}`. No LangGraph: one structured-output call (`RouteDecision{intent, clarifying_question}`, `method=STRUCTURED_OUTPUT_METHOD`) over the coordinator prompt + enumerated intents + the full (summarized) transcript. Intent matched case-insensitively; unmatched/null → clarifying question (constant fallback if the model gave none). The opening turn (no user message) always clarifies. +- Worker dispatch (`services/chatbot/session.py`): all settings — the coordinator's and every route target's (`_resolve_coordinator_routes`) — are resolved in the first DB session, before any LLM call. Routed → the target onboarding agent replies in the same turn (silent handoff, 3 sequential LLM calls: routing + extraction + reply); `active_node_id` → target, or back to the coordinator when the target's turn is complete. Clarify → coordinator replies, `active_node_id` unchanged. An onboarding node behind a coordinator (`_find_return_coordinator`) also returns the session to it on completion; entry-onboarding flows keep the old behaviour. +- `completion_status` is per sub-flow in coordinator flows: `complete` on the finishing turn, `in_progress` again on the next. `collected_fields` is one merged session-wide map, and every agent sees the whole transcript, including other sub-flows' answers. + ## Services / CRUD - `services/chatbot/flow.py` — `get_chatbot_flow`, `save_chatbot_flow_draft`, `publish_chatbot_flow` (row-lock the chatbot via `get_chatbot(for_update=True)` to serialize draft saves and publishes), `validate_flow_for_publish`, `resolve_entry_node`, `find_flow_node`, `parse_flow_nodes/edges`, `dump_flow_nodes/edges`, `build_chatbot_flow_version_public` -- `services/chatbot/session.py` — `start_chatbot_session` (live version + entry node → `create_chatbot_session`), `get_latest_chatbot_session_summary`, `process_chatbot_session_message` (Celery worker body: load session + pinned flow version + active node; inactive session, missing node or a non-onboarding node → failure webhook; otherwise run the turn with the node prompt/config and stored `collected_fields`, store the reply with `node_id`, update progress, POST the webhook) +- `services/chatbot/session.py` — `start_chatbot_session` (live version + entry node → `create_chatbot_session`), `get_latest_chatbot_session_summary`, `process_chatbot_session_message` (Celery worker body: load session + pinned flow version + active node; inactive session, missing node or a non-runnable node (`RUNNABLE_NODE_TYPES` = onboarding, coordinator) → failure webhook; otherwise run the coordinator or onboarding turn with stored `collected_fields`, store the reply with `node_id` = replying node, update progress + `active_node_id`, POST the webhook) - `services/chatbot/agents/onboarding/{graph,prompts,types}.py` — `run_chatbot_turn`, the extraction/reply nodes, `FieldExtraction` +- `services/chatbot/agents/coordinator/{graph,prompts}.py` — `run_coordinator_turn`, `RouteDecision`, `CoordinatorRoute`, `CoordinatorDecision` - `services/chatbot/history.py` — agent-agnostic history plumbing: `summarize_if_needed`, `to_langchain_messages`, `extract_final_reply`, kickoff message, `has_user_turn`, `insert_after_leading_system` - `services/chatbot/types.py` — shared `ChatbotTurnResult`, `ChatbotMessageDict` - `services/chatbot/config.py` — `get_chatbot_config_version` (requested or latest version), `build_chatbot_model_settings` (version row → `ChatbotModelSettings`; 422-class `ChatbotInputError` for non-Anthropic/non-text/no-model configs or a bad thinking budget), `resolve_chatbot_model_settings` (both, used by the worker) -- `services/chatbot/utils.py` — `ChatbotModelSettings`, defaults, `get_chat_model` +- `services/chatbot/utils.py` — `ChatbotModelSettings`, defaults, `get_chat_model`, `STRUCTURED_OUTPUT_METHOD`, `to_provider_error` (Anthropic/client errors → `ChatbotProviderError`, shared by all agents) - `services/chatbot/exceptions.py` — `ChatbotInputError`, `ChatbotProviderError`, `ChatbotSessionConflictError`, `ChatbotNotFoundError`, `ChatbotFlowNotFoundError`, `ChatbotFlowConflictError`, `ChatbotFlowValidationError` -- `crud/chatbot.py` — chatbot: `create_chatbot`, `get_chatbot`, `list_chatbots`, `update_chatbot`, `delete_chatbot`; flow: `get_latest_chatbot_flow_version`, `get_live_chatbot_flow_version`, `get_chatbot_flow_version`, `get_chatbot_flow_version_by_id`, `create_chatbot_flow_version`, `update_chatbot_flow_version_graph`, `publish_chatbot_flow_version` (unsets the previous live before setting the draft live, one commit); session: `create_chatbot_session` (supersedes via `_deactivate_active_chatbot_sessions`), `get_chatbot_session`, `get_latest_chatbot_session` (joined with chatbot + flow version), `update_chatbot_session_progress`, `create_chatbot_session_message`, `list_chatbot_session_messages` +- `crud/chatbot.py` — chatbot: `create_chatbot`, `get_chatbot`, `list_chatbots`, `update_chatbot`, `delete_chatbot`; flow: `get_latest_chatbot_flow_version`, `get_live_chatbot_flow_version`, `get_chatbot_flow_version`, `get_chatbot_flow_version_by_id`, `create_chatbot_flow_version`, `update_chatbot_flow_version_graph`, `publish_chatbot_flow_version` (unsets the previous live before setting the draft live, one commit); session: `create_chatbot_session` (supersedes via `_deactivate_active_chatbot_sessions`), `get_chatbot_session`, `get_latest_chatbot_session` (joined with chatbot + flow version), `update_chatbot_session_progress` (optional `active_node_id`; `None` leaves it unchanged), `create_chatbot_session_message`, `list_chatbot_session_messages` ## Async - `celery/tasks/job_execution.py` — `run_chatbot_session_message` (enqueued via `celery/utils.py:start_chatbot_session_message`) → `process_chatbot_session_message`. - Webhook success payload: `{"success": true, "data": {"session_id", "chatbot_id", "node_id", "reply", "status", "completion_status", "collected_fields"}}`; failure: `{"success": false, "error", "data": {"session_id"}}`. ## External -- Anthropic via `langchain-anthropic`, using the **project's own Anthropic credentials** (`services/chatbot/config.py:get_anthropic_api_key`; no platform `ANTHROPIC_API_KEY` fallback — missing credentials fail publish/first save with 422 and a turn with a failure webhook) — two calls per turn once the user has spoken (extraction + reply), plus one summary call when history exceeds the token budget. +- Anthropic via `langchain-anthropic`, using the **project's own Anthropic credentials** (`services/chatbot/config.py:get_anthropic_api_key`; no platform `ANTHROPIC_API_KEY` fallback — missing credentials fail publish/first save with 422 and a turn with a failure webhook) — two calls per onboarding turn once the user has spoken (extraction + reply); a coordinator turn adds one routing call before a handoff (so 3 on a routed turn, 1 on a clarify turn); plus one summary call when history exceeds the token budget. - Caller webhook (`send_callback`, HMAC-signed with the project webhook secret). - kaapi-frontend React Flow editor reads/writes the flow JSON as-is.