feat(flow): Integrate Rag-based Agentic Flow

backend/app/api/docs/chatbot/create_session.md

@@ -2,4 +2,6 @@ Start a conversation between a chatbot and an end user and return `{session_id,
 
 The session is pinned to the chatbot's live flow version and starts on its entry node (the agent node the `input` node leads to, or the flow's only agent node). Every turn runs that pinned version, so publishing a newer flow applies to new sessions only. `contact_ref` is the caller's own identifier for the end user and is stored as-is.
 
+The entry node's type decides what every turn does: `onboarding` collects the fields its `prompt` asks for; `rag` answers the user's questions from its config's knowledge bases instead, and its `prompt` is the goal that decides whether answers stay within the documents or may use general knowledge. Both node types, and a `rag` node's knowledge bases and OpenAI credentials, are already validated when the flow was published, so a live flow is always runnable — a session is never rejected for a bad node config.
+
 A contact has at most one active session per chatbot. Creating a session marks the contact's previous `active` sessions of this chatbot as `inactive`, in the same transaction; those still `in_progress` become `abandoned`, while `complete` ones stay `complete`. Sessions of other chatbots or other contacts are untouched. Returns 404 if the chatbot doesn't exist in this project or was deleted, 409 if the chatbot has no published flow (publish the flow first) or another session for the same contact was created concurrently (retry).

backend/app/api/docs/chatbot/get_latest_session.md

@@ -2,4 +2,4 @@ Get the most recently created chatbot session for `contact_ref` in the caller's
 
 `contact_ref` is required and matched exactly. Omit `chatbot_id` to get the contact's newest session across the project's chatbots — the response's `chatbot_id` and `chatbot_name` say which one. Sessions of deleted chatbots are not returned.
 
-The response carries the pinned `flow_version`, the `active_node_id` and that node's `prompt`, the session `status` (`active`, or `inactive` once a newer session for the same contact and chatbot was created or the chatbot was deleted), its `completion_status` (`in_progress` until every field the prompt asks for is collected, then `complete`; `abandoned` if it was deactivated while still `in_progress`), and the flat `collected_fields` map (`{field_name: value}`) gathered so far. Returns 404 when no matching session exists in this project.
+The response carries the pinned `flow_version`, the `active_node_id` and that node's `prompt`, the session `status` (`active`, or `inactive` once a newer session for the same contact and chatbot was created or the chatbot was deleted), its `completion_status` (`in_progress` until every field the prompt asks for is collected, then `complete`; `abandoned` if it was deactivated while still `in_progress`), and the flat `collected_fields` map (`{field_name: value}`) gathered so far. A `rag` session's active node doesn't collect fields, so it stays `in_progress` with an empty map. Returns 404 when no matching session exists in this project.

backend/app/api/docs/chatbot/send_session_message.md

@@ -1,7 +1,7 @@
 Queue one turn of a chatbot session; the bot reply is delivered asynchronously to `webhook_url`.
 
-The user `message` is stored immediately (tagged with the session's active node) and the response is `202` with `status: "queued"`. Once the turn completes, Kaapi POSTs `{"success": true, "data": {"session_id", "chatbot_id", "node_id", "reply", "status", "completion_status", "collected_fields"}}` to `webhook_url` (HMAC-signed when the project has a webhook secret). On a model, config or flow failure it POSTs `{"success": false, "error": "...", "data": {"session_id"}}` instead.
+The user `message` is stored immediately (tagged with the session's active node) and the response is `202` with `status: "queued"`. Once the turn completes, Kaapi POSTs `{"success": true, "data": {"session_id", "chatbot_id", "node_id", "reply", "status", "completion_status", "collected_fields", "sources"}}` to `webhook_url` (HMAC-signed when the project has a webhook secret). On a model, config, or knowledge-base search failure it POSTs `{"success": false, "error": "...", "data": {"session_id"}}` instead.
 
-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.
+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. For an `onboarding` node, 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. For a `rag` node, the bot searches the config's knowledge bases when the question may be covered by them and answers from the retrieved passages; `collected_fields` is left unchanged and `completion_status` stays `in_progress`. `sources` lists the documents retrieved for the turn as `[{"file_id", "filename"}]`, deduplicated by `file_id` (always `[]` for `onboarding`, and for `rag` turns that didn't search). `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.
 
 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.

backend/app/services/chatbot/agents/onboarding/graph.py

@@ -4,7 +4,6 @@
 import json
 from typing import Any
 
-import anthropic
 from langchain_core.exceptions import OutputParserException
 from langchain_core.language_models import LanguageModelInput
 from langchain_core.language_models.chat_models import BaseChatModel
@@ -28,14 +27,16 @@
     FieldExtraction,
     OnboardingState,
 )
-from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError
+from app.services.chatbot.exceptions import ChatbotProviderError
 from app.services.chatbot.history import (
     extract_final_reply,
     has_user_turn,
     insert_after_leading_system,
     summarize_if_needed,
     to_langchain_messages,
+    validate_turn_input,
 )
+from app.services.chatbot.provider_errors import CHAT_MODEL_ERRORS, to_provider_error
 from app.services.chatbot.types import ChatbotMessageDict, ChatbotTurnResult
 from app.services.chatbot.utils import ChatbotModelSettings, get_chat_model
 
@@ -166,28 +167,6 @@ 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)"
-    )
-
-
 async def run_chatbot_turn(
     *,
     system_prompt: str,
@@ -197,11 +176,7 @@ async def run_chatbot_turn(
     collected_fields: dict[str, Any] | None = None,
 ) -> ChatbotTurnResult:
     """Run one stateless turn: update the collected fields, then reply."""
-    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."
-        )
+    validate_turn_input(history=history, user_message=user_message)
 
     messages = to_langchain_messages(
         system_prompt=system_prompt, history=history, user_message=user_message
@@ -222,8 +197,8 @@ 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
+    except CHAT_MODEL_ERRORS as e:
+        raise to_provider_error(e, model=model_settings.model) from e
 
     return ChatbotTurnResult(
         reply=extract_final_reply(result["messages"]),

backend/app/services/chatbot/agents/rag/graph.py

@@ -0,0 +1,179 @@
+"""RAG turn: the model answers from the knowledge base, searching it through a tool loop."""
+
+import asyncio
+from typing import Any
+
+from langchain_anthropic import ChatAnthropic
+from langchain_core.language_models import LanguageModelInput
+from langchain_core.messages import (
+    AIMessage,
+    BaseMessage,
+    RemoveMessage,
+    ToolMessage,
+)
+from langchain_core.runnables import Runnable
+from langchain_core.tools import BaseTool
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import REMOVE_ALL_MESSAGES
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.prebuilt.tool_node import ToolInvocationError
+from openai import OpenAI
+
+from app.services.chatbot.agents.rag.prompts import (
+    CONFIG_INSTRUCTIONS_TEMPLATE,
+    GOAL_TEMPLATE,
+    RAG_INSTRUCTION,
+)
+from app.services.chatbot.agents.rag.retriever import build_search_tool
+from app.services.chatbot.agents.rag.types import KnowledgeBasePassage, RagState
+from app.services.chatbot.history import (
+    extract_final_reply,
+    summarize_if_needed,
+    to_langchain_messages,
+    validate_turn_input,
+)
+from app.services.chatbot.provider_errors import CHAT_MODEL_ERRORS, to_provider_error
+from app.services.chatbot.types import (
+    ChatbotMessageDict,
+    ChatbotSource,
+    ChatbotTurnResult,
+)
+from app.services.chatbot.utils import ChatbotModelSettings, get_chat_model
+
+ToolCallingModel = Runnable[LanguageModelInput, BaseMessage]
+
+PREPARE_CONTEXT_NODE = "prepare_context_node"
+AGENT_NODE = "agent_node"
+# tools_condition routes to this exact node name.
+TOOLS_NODE = "tools"
+
+MAX_TOOL_ROUNDS = 3
+# Forced tool use is rejected with thinking on, so the model picks; "none" makes it
+# answer. Only ChatAnthropic passes a dict tool_choice through as-is.
+TOOL_CHOICE_NONE: dict[str, str] = {"type": "none"}
+
+
+def build_rag_system_prompt(*, goal: str, instructions: str | None) -> str:
+    system_prompt = RAG_INSTRUCTION
+    if instructions:
+        system_prompt += CONFIG_INSTRUCTIONS_TEMPLATE.format(instructions=instructions)
+    system_prompt += GOAL_TEMPLATE.format(goal=goal)
+    return system_prompt
+
+
+def _count_tool_rounds(messages: list[BaseMessage]) -> int:
+    # Stored history is plain text, so every tool call in state belongs to this turn.
+    rounds = 0
+    for message in messages:
+        if isinstance(message, AIMessage) and message.tool_calls:
+            rounds += 1
+    return rounds
+
+
+def build_rag_graph(
+    chat_model: ChatAnthropic,
+    search_tool: BaseTool,
+    *,
+    history_token_limit: int,
+) -> CompiledStateGraph:
+    """prepare_context_node -> agent_node <-> tools, ending once the model stops calling tools."""
+    tool_model: ToolCallingModel = chat_model.bind_tools([search_tool])
+    # Tools stay bound so the prior tool_use blocks in the history remain valid.
+    answer_model: ToolCallingModel = chat_model.bind_tools(
+        [search_tool], tool_choice=TOOL_CHOICE_NONE
+    )
+
+    def prepare_context_node(state: RagState) -> dict[str, list[BaseMessage]]:
+        # Summarized once up front so the tool loop never splits a tool_use/tool_result pair.
+        messages = state["messages"]
+        context_view = summarize_if_needed(
+            chat_model, messages, history_token_limit=history_token_limit
+        )
+        if context_view is messages:
+            return {}
+        return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *context_view]}
+
+    def agent_node(state: RagState) -> dict[str, list[BaseMessage]]:
+        messages = state["messages"]
+        model = tool_model
+        if _count_tool_rounds(messages) >= MAX_TOOL_ROUNDS:
+            model = answer_model
+        response = model.invoke(messages)
+        return {"messages": [response]}
+
+    # Retrieval failures must fail the turn (and send a failure webhook), not be fed
+    # back to the model as a tool result; only bad tool arguments are handed back.
+    tool_node = ToolNode([search_tool], handle_tool_errors=ToolInvocationError)
+
+    graph = StateGraph(RagState)
+    graph.add_node(PREPARE_CONTEXT_NODE, prepare_context_node)
+    graph.add_node(AGENT_NODE, agent_node)
+    graph.add_node(TOOLS_NODE, tool_node)
+    graph.add_edge(START, PREPARE_CONTEXT_NODE)
+    graph.add_edge(PREPARE_CONTEXT_NODE, AGENT_NODE)
+    graph.add_conditional_edges(
+        AGENT_NODE, tools_condition, {TOOLS_NODE: TOOLS_NODE, END: END}
+    )
+    graph.add_edge(TOOLS_NODE, AGENT_NODE)
+    return graph.compile()
+
+
+def _collect_sources(messages: list[BaseMessage]) -> list[ChatbotSource]:
+    sources: list[ChatbotSource] = []
+    seen_file_ids: set[str] = set()
+    for message in messages:
+        if not isinstance(message, ToolMessage) or not message.artifact:
+            continue
+        passages: list[KnowledgeBasePassage] = message.artifact
+        for passage in passages:
+            if passage.file_id in seen_file_ids:
+                continue
+            seen_file_ids.add(passage.file_id)
+            sources.append({"file_id": passage.file_id, "filename": passage.filename})
+    return sources
+
+
+async def run_rag_turn(
+    *,
+    system_prompt: str,
+    history: list[ChatbotMessageDict],
+    user_message: str | None,
+    model_settings: ChatbotModelSettings,
+    openai_client: OpenAI,
+    collected_fields: dict[str, Any] | None = None,
+) -> ChatbotTurnResult:
+    """Run one stateless turn: answer the user from the configured knowledge bases."""
+    validate_turn_input(history=history, user_message=user_message)
+
+    rag_system_prompt = build_rag_system_prompt(
+        goal=system_prompt, instructions=model_settings.instructions
+    )
+    initial_state: RagState = {
+        "messages": to_langchain_messages(
+            system_prompt=rag_system_prompt, history=history, user_message=user_message
+        ),
+    }
+    search_tool = build_search_tool(
+        client=openai_client,
+        knowledge_base_ids=model_settings.knowledge_base_ids,
+        max_num_results=model_settings.max_num_results,
+    )
+    try:
+        graph = build_rag_graph(
+            get_chat_model(model_settings),
+            search_tool,
+            history_token_limit=model_settings.history_token_limit,
+        )
+        # ChatAnthropic.invoke and the OpenAI search are sync; keep them off the event loop.
+        result = await asyncio.to_thread(graph.invoke, initial_state)
+    except CHAT_MODEL_ERRORS as e:
+        raise to_provider_error(e, model=model_settings.model) from e
+
+    final_messages: list[BaseMessage] = result["messages"]
+    return ChatbotTurnResult(
+        reply=extract_final_reply(final_messages),
+        collected_fields=dict(collected_fields) if collected_fields else {},
+        is_complete=False,
+        sources=_collect_sources(final_messages),
+    )

backend/app/services/chatbot/agents/rag/prompts.py

@@ -0,0 +1,35 @@
+SEARCH_TOOL_NAME = "search_knowledge_base"
+SEARCH_TOOL_DESCRIPTION = (
+    "Search the knowledge base documents and return the most relevant passages. "
+    "Call it whenever the user's question may be answered by the documents."
+)
+SEARCH_QUERY_DESCRIPTION = (
+    "A standalone search query: resolve pronouns and references from the "
+    "conversation so the query makes sense without it."
+)
+
+NO_PASSAGES_FOUND = (
+    "No relevant passages were found in the knowledge base for this query."
+)
+PASSAGE_TEMPLATE = '<source file="{filename}">\n{text}\n</source>'
+PASSAGE_SEPARATOR = "\n\n"
+
+RAG_INSTRUCTION = (
+    "You are an assistant that answers the user's questions using a knowledge base "
+    f"of documents, searchable with the `{SEARCH_TOOL_NAME}` tool.\n"
+    f"1. Call `{SEARCH_TOOL_NAME}` when the question may be covered by the documents; "
+    "skip it for greetings and small talk.\n"
+    "2. Write each search query as a standalone question, resolving references from "
+    "the conversation.\n"
+    "3. Ground your answers in the retrieved passages and never invent facts. If the "
+    "passages don't cover the question, say so.\n"
+    "4. The goal decides your scope. If it restricts answers to the knowledge base, "
+    "politely decline anything the documents don't cover. Otherwise you may use "
+    "general knowledge, but say that part isn't from the documents.\n"
+    "5. If the goal and the configuration instructions conflict, follow the goal.\n"
+    "6. Never mention tools, searches, passages, file names, or ids to the user."
+)
+CONFIG_INSTRUCTIONS_TEMPLATE = (
+    "\n\n<config_instructions>\n{instructions}\n</config_instructions>"
+)
+GOAL_TEMPLATE = "\n\n<goal>\n{goal}\n</goal>"