diff --git a/backend/app/api/docs/chatbot/create_session.md b/backend/app/api/docs/chatbot/create_session.md index 90a640e80..24ca230ab 100644 --- a/backend/app/api/docs/chatbot/create_session.md +++ b/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). diff --git a/backend/app/api/docs/chatbot/get_latest_session.md b/backend/app/api/docs/chatbot/get_latest_session.md index 20591ec3e..17a7f2001 100644 --- a/backend/app/api/docs/chatbot/get_latest_session.md +++ b/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. diff --git a/backend/app/api/docs/chatbot/send_session_message.md b/backend/app/api/docs/chatbot/send_session_message.md index e6d1034d0..69b20507b 100644 --- a/backend/app/api/docs/chatbot/send_session_message.md +++ b/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. diff --git a/backend/app/services/chatbot/agents/onboarding/graph.py b/backend/app/services/chatbot/agents/onboarding/graph.py index 4c8f32ded..4e2efd525 100644 --- a/backend/app/services/chatbot/agents/onboarding/graph.py +++ b/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"]), diff --git a/backend/app/services/chatbot/agents/rag/__init__.py b/backend/app/services/chatbot/agents/rag/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/services/chatbot/agents/rag/graph.py b/backend/app/services/chatbot/agents/rag/graph.py new file mode 100644 index 000000000..2a390af3e --- /dev/null +++ b/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), + ) diff --git a/backend/app/services/chatbot/agents/rag/prompts.py b/backend/app/services/chatbot/agents/rag/prompts.py new file mode 100644 index 000000000..4a484f66e --- /dev/null +++ b/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 = '\n{text}\n' +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\n{instructions}\n" +) +GOAL_TEMPLATE = "\n\n\n{goal}\n" diff --git a/backend/app/services/chatbot/agents/rag/retriever.py b/backend/app/services/chatbot/agents/rag/retriever.py new file mode 100644 index 000000000..6d937bdce --- /dev/null +++ b/backend/app/services/chatbot/agents/rag/retriever.py @@ -0,0 +1,216 @@ +"""Knowledge-base search over OpenAI vector stores; Anthropic has no native file search.""" + +import html +import logging + +import openai +from langchain_core.tools import BaseTool, StructuredTool +from openai import OpenAI +from sqlmodel import Session + +from app.crud import get_provider_credential +from app.crud.rag.open_ai import OPENAI_TIMEOUT_SECONDS +from app.models.llm.constants import Provider +from app.services.chatbot.agents.rag.prompts import ( + NO_PASSAGES_FOUND, + PASSAGE_SEPARATOR, + PASSAGE_TEMPLATE, + SEARCH_TOOL_DESCRIPTION, + SEARCH_TOOL_NAME, +) +from app.services.chatbot.agents.rag.types import ( + KnowledgeBasePassage, + SearchKnowledgeBaseArgs, +) +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.services.chatbot.provider_errors import PROVIDER_SERVER_ERROR_STATUS +from app.utils import mask_string + +logger = logging.getLogger(__name__) + +DEFAULT_MAX_NUM_RESULTS = 5 +OPENAI_MAX_NUM_RESULTS = 50 +# A search is an idempotent read, so one SDK retry absorbs a transient blip +# without stretching the turn far past OPENAI_TIMEOUT_SECONDS. +KB_SEARCH_MAX_RETRIES = 1 +CREDENTIAL_API_KEY = "api_key" +PASSAGE_CHUNK_SEPARATOR = "\n" + + +def get_knowledge_base_client( + *, session: Session, organization_id: int, project_id: int +) -> OpenAI: + """OpenAI client from the project's credentials, used only for vector-store search.""" + credentials = get_provider_credential( + session=session, + org_id=organization_id, + project_id=project_id, + provider=Provider.OPENAI.value, + ) + if not isinstance(credentials, dict) or not credentials.get(CREDENTIAL_API_KEY): + error_message = ( + "[KAAPI] The rag agent searches knowledge bases with the project's OpenAI " + "credentials, but none are configured; add OpenAI credentials to this project." + ) + logger.warning( + f"[get_knowledge_base_client] {error_message} | project_id: {project_id}" + ) + raise ChatbotInputError(error_message) + + api_key = credentials[CREDENTIAL_API_KEY] + logger.info( + f"[get_knowledge_base_client] OpenAI client configured | " + f"project_id: {project_id}, api_key: {mask_string(api_key)}" + ) + return OpenAI( + api_key=api_key, + timeout=OPENAI_TIMEOUT_SECONDS, + max_retries=KB_SEARCH_MAX_RETRIES, + ) + + +def _resolve_top_k(max_num_results: int | None) -> int: + if max_num_results is None: + return DEFAULT_MAX_NUM_RESULTS + return min(max_num_results, OPENAI_MAX_NUM_RESULTS) + + +def _to_search_error(error: openai.OpenAIError, *, vector_store_id: str) -> str: + # APITimeoutError subclasses APIConnectionError, so it must be checked first. + if isinstance(error, openai.APITimeoutError): + error_message = ( + "[KAAPI] Knowledge base search timed out — retry the turn. If persistent, " + "contact Kaapi. (code: APITimeoutError)" + ) + logger.error( + f"[search_knowledge_bases] {error_message} | provider=openai, " + f"vector_store_id: {vector_store_id}", + exc_info=error, + ) + return error_message + if isinstance(error, openai.APIConnectionError): + error_message = ( + "[KAAPI] Network/DNS issue reaching OpenAI for knowledge base search — " + "check connectivity. If persistent, contact Kaapi. (code: APIConnectionError)" + ) + logger.error( + f"[search_knowledge_bases] {error_message} | provider=openai, " + f"vector_store_id: {vector_store_id}", + exc_info=error, + ) + return error_message + if isinstance(error, openai.APIStatusError): + error_message = ( + f"[OPENAI] Knowledge base search failed for '{vector_store_id}': " + f"{error.message} (code: {error.status_code}, request_id: {error.request_id})" + ) + # 5xx is provider-side (alert-worthy); 4xx is caller's fault (noise if alerted) + log = ( + logger.error + if error.status_code >= PROVIDER_SERVER_ERROR_STATUS + else logger.warning + ) + log( + f"[search_knowledge_bases] {error_message} | provider=openai, " + f"vector_store_id: {vector_store_id}", + exc_info=error, + ) + return error_message + + error_message = ( + f"[OPENAI] Knowledge base search failed for '{vector_store_id}': {error} " + f"— retry the turn. If persistent, contact Kaapi. (code: {type(error).__name__})" + ) + logger.error( + f"[search_knowledge_bases] {error_message} | provider=openai, " + f"vector_store_id: {vector_store_id}", + exc_info=error, + ) + return error_message + + +def search_knowledge_bases( + *, + client: OpenAI, + knowledge_base_ids: tuple[str, ...], + query: str, + max_num_results: int | None, +) -> list[KnowledgeBasePassage]: + """Search every knowledge base and keep the overall top passages by score.""" + top_k = _resolve_top_k(max_num_results) + passages: list[KnowledgeBasePassage] = [] + for vector_store_id in knowledge_base_ids: + logger.info( + f"[search_knowledge_bases] Searching knowledge base | " + f"vector_store_id: {vector_store_id}, max_num_results: {top_k}" + ) + try: + page = client.vector_stores.search( + vector_store_id, query=query, max_num_results=top_k + ) + except openai.OpenAIError as e: + error_message = _to_search_error(e, vector_store_id=vector_store_id) + raise ChatbotProviderError(error_message) from e + + for result in page.data: + chunk_texts: list[str] = [] + for chunk in result.content: + chunk_texts.append(chunk.text) + passages.append( + KnowledgeBasePassage( + file_id=result.file_id, + filename=result.filename, + score=result.score, + text=PASSAGE_CHUNK_SEPARATOR.join(chunk_texts), + ) + ) + + passages.sort(key=lambda passage: passage.score, reverse=True) + top_passages = passages[:top_k] + logger.info( + f"[search_knowledge_bases] Knowledge base search finished | " + f"knowledge_bases: {len(knowledge_base_ids)}, passages: {len(top_passages)}" + ) + return top_passages + + +def format_passages(passages: list[KnowledgeBasePassage]) -> str: + if not passages: + return NO_PASSAGES_FOUND + + rendered: list[str] = [] + for passage in passages: + rendered.append( + PASSAGE_TEMPLATE.format( + filename=html.escape(passage.filename, quote=True), text=passage.text + ) + ) + return PASSAGE_SEPARATOR.join(rendered) + + +def build_search_tool( + *, + client: OpenAI, + knowledge_base_ids: tuple[str, ...], + max_num_results: int | None, +) -> BaseTool: + """`search_knowledge_base` tool: passages as content, the structured hits as artifact.""" + + def search_knowledge_base( + query: str, + ) -> tuple[str, list[KnowledgeBasePassage]]: + passages = search_knowledge_bases( + client=client, + knowledge_base_ids=knowledge_base_ids, + query=query, + max_num_results=max_num_results, + ) + return format_passages(passages), passages + + return StructuredTool.from_function( + func=search_knowledge_base, + name=SEARCH_TOOL_NAME, + description=SEARCH_TOOL_DESCRIPTION, + args_schema=SearchKnowledgeBaseArgs, + response_format="content_and_artifact", + ) diff --git a/backend/app/services/chatbot/agents/rag/types.py b/backend/app/services/chatbot/agents/rag/types.py new file mode 100644 index 000000000..a1b85ec87 --- /dev/null +++ b/backend/app/services/chatbot/agents/rag/types.py @@ -0,0 +1,24 @@ +from dataclasses import dataclass +from typing import Annotated, TypedDict + +from langchain_core.messages import BaseMessage +from langgraph.graph.message import add_messages +from pydantic import BaseModel, Field + +from app.services.chatbot.agents.rag.prompts import SEARCH_QUERY_DESCRIPTION + + +class RagState(TypedDict): + messages: Annotated[list[BaseMessage], add_messages] + + +@dataclass(frozen=True) +class KnowledgeBasePassage: + file_id: str + filename: str + score: float + text: str + + +class SearchKnowledgeBaseArgs(BaseModel): + query: str = Field(min_length=1, description=SEARCH_QUERY_DESCRIPTION) diff --git a/backend/app/services/chatbot/agents/registry.py b/backend/app/services/chatbot/agents/registry.py new file mode 100644 index 000000000..ace4e70aa --- /dev/null +++ b/backend/app/services/chatbot/agents/registry.py @@ -0,0 +1,55 @@ +import logging +from typing import Any + +from openai import OpenAI + +from app.models.chatbot import ChatbotFlowNodeTypeEnum +from app.services.chatbot.agents.onboarding.graph import run_chatbot_turn +from app.services.chatbot.agents.rag.graph import run_rag_turn +from app.services.chatbot.exceptions import ChatbotInputError +from app.services.chatbot.types import ChatbotMessageDict, ChatbotTurnResult +from app.services.chatbot.utils import ChatbotModelSettings + +logger = logging.getLogger(__name__) + + +async def run_agent_turn( + *, + node_type: ChatbotFlowNodeTypeEnum, + system_prompt: str, + history: list[ChatbotMessageDict], + user_message: str | None, + model_settings: ChatbotModelSettings, + collected_fields: dict[str, Any] | None = None, + openai_client: OpenAI | None = None, +) -> ChatbotTurnResult: + """Run one turn with the agent the session's active flow node runs.""" + if node_type == ChatbotFlowNodeTypeEnum.ONBOARDING: + return await run_chatbot_turn( + system_prompt=system_prompt, + history=history, + user_message=user_message, + model_settings=model_settings, + collected_fields=collected_fields, + ) + + if node_type == ChatbotFlowNodeTypeEnum.RAG: + if openai_client is None: + error_message = ( + "[KAAPI] The rag agent needs an OpenAI client to search its " + "knowledge bases." + ) + logger.error(f"[run_agent_turn] {error_message}") + raise ChatbotInputError(error_message) + return await run_rag_turn( + system_prompt=system_prompt, + history=history, + user_message=user_message, + model_settings=model_settings, + openai_client=openai_client, + collected_fields=collected_fields, + ) + + error_message = f"[KAAPI] Unsupported chatbot node type '{node_type}'." + logger.error(f"[run_agent_turn] {error_message}") + raise ChatbotInputError(error_message) diff --git a/backend/app/services/chatbot/config.py b/backend/app/services/chatbot/config.py index 30b4e74ab..071fcff20 100644 --- a/backend/app/services/chatbot/config.py +++ b/backend/app/services/chatbot/config.py @@ -1,16 +1,20 @@ +import logging from dataclasses import replace from typing import Any, get_args from uuid import UUID from fastapi import HTTPException +from openai import OpenAI from pydantic import ValidationError from sqlmodel import Session +from app.crud import CollectionCrud from app.crud.config import ConfigVersionCrud from app.crud.credentials import get_provider_credential -from app.models import ConfigVersion +from app.models import ConfigVersion, ProviderType from app.models.llm.constants import Provider from app.models.llm.request import ConfigBlob, KaapiTextCompletionConfig +from app.services.chatbot.agents.rag.retriever import get_knowledge_base_client from app.services.chatbot.exceptions import ChatbotInputError from app.services.chatbot.utils import ( DEFAULT_HISTORY_TOKEN_LIMIT, @@ -20,6 +24,8 @@ default_thinking, ) +logger = logging.getLogger(__name__) + _ANTHROPIC_EFFORT_LEVELS: tuple[AnthropicEffort, ...] = get_args(AnthropicEffort) THINKING_TYPE_ENABLED = "enabled" CREDENTIAL_API_KEY = "api_key" @@ -158,5 +164,60 @@ def build_chatbot_model_settings( if history_token_limit is not None else DEFAULT_HISTORY_TOKEN_LIMIT ), + instructions=params.instructions, + knowledge_base_ids=tuple(params.knowledge_base_ids or ()), + max_num_results=params.max_num_results, ) return model_settings + + +def _validate_knowledge_base_ids( + *, session: Session, project_id: int, knowledge_base_ids: tuple[str, ...] +) -> None: + if not knowledge_base_ids: + error_message = ( + "[KAAPI] The rag agent needs a config with non-empty " + "completion.params.knowledge_base_ids; add the collection's knowledge base id." + ) + logger.warning( + f"[_validate_knowledge_base_ids] {error_message} | project_id: {project_id}" + ) + raise ChatbotInputError(error_message) + + known_ids: set[str] = set() + for collection in CollectionCrud(session, project_id).read_all(): + if collection.provider == ProviderType.openai: + known_ids.add(collection.knowledge_base_id) + + unknown_ids: list[str] = [] + for knowledge_base_id in knowledge_base_ids: + if knowledge_base_id not in known_ids: + unknown_ids.append(knowledge_base_id) + if unknown_ids: + error_message = ( + "[KAAPI] Config knowledge_base_ids not found among this project's OpenAI " + f"collections: {', '.join(unknown_ids)}; use the knowledge_base_id of an " + "existing collection." + ) + logger.warning( + f"[_validate_knowledge_base_ids] {error_message} | project_id: {project_id}" + ) + raise ChatbotInputError(error_message) + + +def prepare_rag_search_client( + *, + session: Session, + organization_id: int, + project_id: int, + model_settings: ChatbotModelSettings, +) -> OpenAI: + """Validate the rag agent's knowledge bases and return the client that searches them.""" + _validate_knowledge_base_ids( + session=session, + project_id=project_id, + knowledge_base_ids=model_settings.knowledge_base_ids, + ) + return get_knowledge_base_client( + session=session, organization_id=organization_id, project_id=project_id + ) diff --git a/backend/app/services/chatbot/flow.py b/backend/app/services/chatbot/flow.py index b5f78bac9..924ce21c2 100644 --- a/backend/app/services/chatbot/flow.py +++ b/backend/app/services/chatbot/flow.py @@ -28,6 +28,7 @@ build_chatbot_model_settings, get_anthropic_api_key, get_chatbot_config_version, + prepare_rag_search_client, ) from app.services.chatbot.exceptions import ( ChatbotFlowConflictError, @@ -41,9 +42,9 @@ 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. +# The runtime executes onboarding and rag agents; coordinator is stored but not publishable yet. SUPPORTED_AGENT_NODE_TYPES: frozenset[ChatbotFlowNodeTypeEnum] = frozenset( - {ChatbotFlowNodeTypeEnum.ONBOARDING} + {ChatbotFlowNodeTypeEnum.ONBOARDING, ChatbotFlowNodeTypeEnum.RAG} ) _SUPPORTED_AGENT_NODE_TYPES_LABEL = ", ".join( sorted(node_type.value for node_type in SUPPORTED_AGENT_NODE_TYPES) @@ -207,7 +208,7 @@ def _list_structural_problems( def _pin_agent_node_config( - *, session: Session, project_id: int, node: ChatbotFlowNode + *, session: Session, organization_id: int, project_id: int, node: ChatbotFlowNode ) -> tuple[ChatbotFlowNode, list[str]]: """Check the agent node is runnable; returns the node with its config_version pinned, plus any problems.""" if node.type not in SUPPORTED_AGENT_NODE_TYPES: @@ -236,7 +237,7 @@ def _pin_agent_node_config( return node, problems try: - build_chatbot_model_settings( + model_settings = build_chatbot_model_settings( version_row, max_tokens=None, history_token_limit=None ) except ChatbotInputError as e: @@ -244,6 +245,19 @@ def _pin_agent_node_config( problems.append(f"node '{node.id}': {reason}") return node, problems + if node.type == ChatbotFlowNodeTypeEnum.RAG: + try: + prepare_rag_search_client( + session=session, + organization_id=organization_id, + project_id=project_id, + model_settings=model_settings, + ) + except ChatbotInputError as e: + reason = str(e).removeprefix(KAAPI_ERROR_TAG).strip() + problems.append(f"node '{node.id}': {reason}") + return node, problems + # Pinned at publish so later config edits never change a published flow. pinned_data = {**node.data, "config_version": version_row.version} return node.model_copy(update={"data": pinned_data}), problems @@ -275,7 +289,10 @@ def validate_flow_for_publish( pinned_nodes.append(node) continue pinned_node, node_problems = _pin_agent_node_config( - session=session, project_id=project_id, node=node + session=session, + organization_id=organization_id, + project_id=project_id, + node=node, ) pinned_nodes.append(pinned_node) problems.extend(node_problems) diff --git a/backend/app/services/chatbot/history.py b/backend/app/services/chatbot/history.py index 2c1d94595..4967489c3 100644 --- a/backend/app/services/chatbot/history.py +++ b/backend/app/services/chatbot/history.py @@ -1,5 +1,7 @@ """Conversation-history plumbing shared by every chatbot agent graph.""" +import logging + from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -8,6 +10,8 @@ from app.services.chatbot.types import ChatbotMessageDict from app.services.chatbot.utils import FLOW_KEEP_RECENT_MESSAGES +logger = logging.getLogger(__name__) + SUMMARY_NOTE_PREFIX = "[Earlier conversation summary] " SUMMARIZE_INSTRUCTION = ( "Summarize the conversation so far in 2-3 sentences. Keep every fact the " @@ -122,5 +126,16 @@ def extract_final_reply(messages: list[BaseMessage]) -> str: "[ANTHROPIC] Chat model returned no reply text — retry the turn. " "If persistent, contact Kaapi. (code: EmptyReply)" ) + logger.warning(f"[extract_final_reply] {error_message}") raise ChatbotProviderError(error_message) return reply + + +def validate_turn_input( + *, history: list[ChatbotMessageDict], user_message: str | None +) -> None: + 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." + ) diff --git a/backend/app/services/chatbot/provider_errors.py b/backend/app/services/chatbot/provider_errors.py new file mode 100644 index 000000000..7430c8145 --- /dev/null +++ b/backend/app/services/chatbot/provider_errors.py @@ -0,0 +1,68 @@ +"""Anthropic chat-model error mapping shared by every chatbot agent.""" + +import logging + +import anthropic + +from app.services.chatbot.exceptions import ChatbotProviderError + +logger = logging.getLogger(__name__) + +PROVIDER_SERVER_ERROR_STATUS = 500 + +# ValueError covers langchain_anthropic's client-side rejection of a model config +# (e.g. thinking/effort the model doesn't support) before any request is sent. +CHAT_MODEL_ERRORS: tuple[type[Exception], ...] = ( + anthropic.APIConnectionError, + anthropic.APIStatusError, + ValueError, +) + + +def to_provider_error(error: Exception, *, model: str) -> ChatbotProviderError: + # APITimeoutError subclasses APIConnectionError, so it must be checked first. + if isinstance(error, anthropic.APITimeoutError): + error_message = ( + "[KAAPI] Request timed out — retry smaller. If persistent, contact Kaapi. " + "(code: APITimeoutError)" + ) + logger.error( + f"[to_provider_error] {error_message} | provider=anthropic, model={model}", + exc_info=error, + ) + return ChatbotProviderError(error_message) + if isinstance(error, anthropic.APIConnectionError): + error_message = ( + "[KAAPI] Network/DNS issue reaching provider — check connectivity. " + "If persistent, contact Kaapi. (code: APIConnectionError)" + ) + logger.error( + f"[to_provider_error] {error_message} | provider=anthropic, model={model}", + exc_info=error, + ) + return ChatbotProviderError(error_message) + if isinstance(error, anthropic.APIStatusError): + error_message = ( + f"[ANTHROPIC] Chat model call failed: {error.message} " + f"(code: {error.status_code}, request_id: {error.request_id})" + ) + # 5xx is provider-side (alert-worthy); 4xx is caller's fault (noise if alerted) + log = ( + logger.error + if error.status_code >= PROVIDER_SERVER_ERROR_STATUS + else logger.warning + ) + log( + f"[to_provider_error] {error_message} | provider=anthropic, model={model}", + exc_info=error, + ) + return ChatbotProviderError(error_message) + + error_message = ( + f"[KAAPI] Unsupported model configuration for '{model}': {error} " + "— update the config's model, thinking, or effort. (code: ValueError)" + ) + logger.warning( + f"[to_provider_error] {error_message} | provider=anthropic, model={model}" + ) + return ChatbotProviderError(error_message) diff --git a/backend/app/services/chatbot/session.py b/backend/app/services/chatbot/session.py index 3f7da127c..dcb62a47e 100644 --- a/backend/app/services/chatbot/session.py +++ b/backend/app/services/chatbot/session.py @@ -3,6 +3,7 @@ from uuid import UUID from fastapi import HTTPException +from openai import OpenAI from sqlmodel import Session from app.core.db import engine @@ -26,14 +27,18 @@ ChatbotSessionStatusEnum, ChatbotSessionSummary, ) -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.agents.registry import run_agent_turn +from app.services.chatbot.config import ( + prepare_rag_search_client, + resolve_chatbot_model_settings, +) from app.services.chatbot.exceptions import ( ChatbotFlowConflictError, ChatbotInputError, ChatbotProviderError, ) from app.services.chatbot.flow import ( + SUPPORTED_AGENT_NODE_TYPES, find_flow_node, get_chatbot_or_raise, parse_flow_edges, @@ -52,6 +57,7 @@ class _TurnNode(NamedTuple): node_id: str + node_type: ChatbotFlowNodeTypeEnum prompt: str config_id: UUID config_version: int | None @@ -136,7 +142,7 @@ def get_latest_chatbot_session_summary( def _resolve_turn_node( *, flow_version: ChatbotFlowVersion | None, active_node_id: str ) -> _TurnNode: - """The runnable onboarding node the session is on; ChatbotInputError when it can't run.""" + """The runnable agent 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( @@ -149,11 +155,13 @@ def _resolve_turn_node( ) raise ChatbotInputError(error_message) - if active_node.type != ChatbotFlowNodeTypeEnum.ONBOARDING: + if active_node.type not in SUPPORTED_AGENT_NODE_TYPES: + supported = ", ".join( + sorted(node_type.value for node_type in SUPPORTED_AGENT_NODE_TYPES) + ) error_message = ( f"[KAAPI] Active node '{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"which the chatbot runtime does not support yet; only {supported} nodes can run." ) raise ChatbotInputError(error_message) @@ -167,6 +175,7 @@ def _resolve_turn_node( return _TurnNode( node_id=active_node.id, + node_type=active_node.type, prompt=node_data.prompt, config_id=node_data.config_id, config_version=node_data.config_version, @@ -187,7 +196,7 @@ def _split_pending_turn( if not history: return history, None if history[-1]["role"] != ChatbotMessageRoleEnum.USER: - # No pending user turn; run_chatbot_turn rejects this with ChatbotInputError. + # No pending user turn; the agent runner rejects this with ChatbotInputError. return history, None pending = history.pop() return history, pending["content"] @@ -251,6 +260,7 @@ def process_chatbot_session_message( flow_version = get_chatbot_flow_version_by_id( session, flow_version_id=chatbot_session.flow_version_id ) + openai_client: OpenAI | None = None try: turn_node = _resolve_turn_node( @@ -266,6 +276,13 @@ def process_chatbot_session_message( max_tokens=None, history_token_limit=None, ) + if turn_node.node_type == ChatbotFlowNodeTypeEnum.RAG: + openai_client = prepare_rag_search_client( + session=session, + organization_id=organization_id, + project_id=project_id, + model_settings=model_settings, + ) except (HTTPException, ChatbotInputError) as e: error = str(e.detail) if isinstance(e, HTTPException) else str(e) _send_failure_webhook( @@ -280,12 +297,14 @@ def process_chatbot_session_message( try: turn_result = asyncio.run( - run_chatbot_turn( + run_agent_turn( + node_type=turn_node.node_type, system_prompt=turn_node.prompt, history=history, user_message=user_message, model_settings=model_settings, collected_fields=collected_fields, + openai_client=openai_client, ) ) except (ChatbotInputError, ChatbotProviderError) as e: @@ -323,6 +342,7 @@ def process_chatbot_session_message( "status": session_status, "completion_status": completion_status.value, "collected_fields": turn_result.collected_fields, + "sources": turn_result.sources, } ) send_callback(webhook_url, envelope.model_dump(exclude_none=True), webhook_secret) diff --git a/backend/app/services/chatbot/types.py b/backend/app/services/chatbot/types.py index bcc08759e..71df0e814 100644 --- a/backend/app/services/chatbot/types.py +++ b/backend/app/services/chatbot/types.py @@ -1,16 +1,22 @@ -from dataclasses import dataclass -from typing import Any +from dataclasses import dataclass, field +from typing import Any, TypedDict from app.models.chatbot import ChatbotCompletionStatusEnum ChatbotMessageDict = dict[str, str] +class ChatbotSource(TypedDict): + file_id: str + filename: str + + @dataclass(frozen=True) class ChatbotTurnResult: reply: str collected_fields: dict[str, Any] is_complete: bool + sources: list[ChatbotSource] = field(default_factory=list) @property def completion_status(self) -> ChatbotCompletionStatusEnum: diff --git a/backend/app/services/chatbot/utils.py b/backend/app/services/chatbot/utils.py index 764f261ca..e65693d05 100644 --- a/backend/app/services/chatbot/utils.py +++ b/backend/app/services/chatbot/utils.py @@ -25,6 +25,11 @@ class ChatbotModelSettings: reasoning_effort: AnthropicEffort | None = None thinking: dict[str, Any] = field(default_factory=default_thinking) history_token_limit: int = DEFAULT_HISTORY_TOKEN_LIMIT + instructions: str | None = None + # OpenAI vector store ids (Collection.knowledge_base_id); only the rag agent reads them. + knowledge_base_ids: tuple[str, ...] = () + max_num_results: int | None = None + # The project's own Anthropic key; kept out of repr/eq so it never lands in logs or test diffs. api_key: str = field(default="", repr=False, compare=False) diff --git a/backend/app/tests/api/routes/chatbot/test_flow.py b/backend/app/tests/api/routes/chatbot/test_flow.py index 06c10b9f8..0356a8cc8 100644 --- a/backend/app/tests/api/routes/chatbot/test_flow.py +++ b/backend/app/tests/api/routes/chatbot/test_flow.py @@ -8,7 +8,7 @@ from httpx import Response from sqlmodel import Session -from app.models import Chatbot, Config +from app.models import Chatbot, Config, Project from app.tests.utils.auth import TestAuthContext from app.tests.utils.chatbot import ( CHATBOTS_URL, @@ -27,6 +27,7 @@ random_contact_ref, sessions_url, ) +from app.tests.utils.collection import get_vector_store_collection from app.tests.utils.test_data import create_test_config INVALID_FLOW_PREFIX = "[KAAPI] Chatbot flow is invalid: " @@ -430,6 +431,33 @@ def test_publishes_draft_and_unsets_previous_live( assert previous.is_live is False assert published.is_live is True + def test_rag_node_with_knowledge_bases_publishes_as_entry_node( + self, + db: Session, + user_api_key: TestAuthContext, + chatbot: Chatbot, + save: Callable[..., Response], + publish: Callable[[], Response], + live_v1: dict[str, Any], + ) -> None: + project = db.get(Project, user_api_key.project_id) + assert project is not None + collection = get_vector_store_collection(db, project) + rag_config = create_test_anthropic_config( + db, + project_id=user_api_key.project_id, + knowledge_base_ids=[collection.knowledge_base_id], + ) + save([build_onboarding_node(config_id=rag_config.id, node_type="rag")]) + + resp = publish() + + assert resp.status_code == 200 + assert resp.json()["data"]["version"] == 2 + _, stored = list_stored_flow_versions(db, chatbot) + assert (stored.version, stored.is_live) == (2, True) + assert stored.nodes[0]["type"] == "rag" + def test_missing_config_version_is_pinned_to_latest_and_stored( self, db: Session, @@ -585,9 +613,10 @@ def test_unknown_chatbot_returns_404( ), f"node '{ONBOARDING_NODE_ID}' has type 'coordinator', which is not supported yet", ), - "rag-node": ( + "rag-node-without-knowledge-bases": ( lambda cid: ([build_onboarding_node(config_id=cid, node_type="rag")], []), - f"node '{ONBOARDING_NODE_ID}' has type 'rag', which is not supported yet", + f"node '{ONBOARDING_NODE_ID}': The rag agent needs a config with non-empty " + "completion.params.knowledge_base_ids", ), "missing-config-id": ( lambda cid: ([build_onboarding_node(config_id=None)], []), diff --git a/backend/app/tests/services/chatbot/agents/onboarding/test_graph.py b/backend/app/tests/services/chatbot/agents/onboarding/test_graph.py index 220cd9d8b..4576807c4 100644 --- a/backend/app/tests/services/chatbot/agents/onboarding/test_graph.py +++ b/backend/app/tests/services/chatbot/agents/onboarding/test_graph.py @@ -64,6 +64,7 @@ async def test_first_call_returns_greeting_without_kickoff( result = await _run_turn(history=[], user_message=None) assert result.reply == "Hi! What's your name?" + assert result.sources == [] # Anthropic rejects a conversation that doesn't open with a user turn. sent = chat_model.calls[0] assert isinstance(sent[0], SystemMessage) @@ -210,7 +211,7 @@ async def test_long_history_over_budget_is_summarized( "bot 3", "user 4", ] - # The reply is the turn's answer, not the summary text. + # The reply is the turn's answer, never the summary text. assert result.reply == "Got it." @pytest.mark.asyncio diff --git a/backend/app/tests/services/chatbot/agents/rag/__init__.py b/backend/app/tests/services/chatbot/agents/rag/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/tests/services/chatbot/agents/rag/test_graph.py b/backend/app/tests/services/chatbot/agents/rag/test_graph.py new file mode 100644 index 000000000..3b2891608 --- /dev/null +++ b/backend/app/tests/services/chatbot/agents/rag/test_graph.py @@ -0,0 +1,319 @@ +import anthropic +import httpx +import openai +import pytest +from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage + +from app.services.chatbot.agents.rag.graph import run_rag_turn +from app.services.chatbot.agents.rag.prompts import RAG_INSTRUCTION +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.services.chatbot.utils import ChatbotModelSettings +from app.tests.utils.chatbot import ( + FakeOpenAIClient, + FakeVectorStores, + ScriptedChatModel, + patch_chat_model, + search_hit, + tool_call_reply, +) + +GOAL = "Answer questions about school fees only from the documents." +SUMMARY_TEXT = "User asked about admissions." + +_ANTHROPIC_REQUEST = httpx.Request("POST", "https://api.anthropic.com/v1/messages") +_OPENAI_REQUEST = httpx.Request("POST", "https://api.openai.com/v1/vector_stores") + + +def _model_settings( + *, instructions: str | None = None, history_token_limit: int = 4000 +) -> ChatbotModelSettings: + return ChatbotModelSettings( + model="claude-sonnet-4-6", + instructions=instructions, + knowledge_base_ids=("vs_fees",), + history_token_limit=history_token_limit, + ) + + +def _fees_client() -> FakeOpenAIClient: + return FakeOpenAIClient( + vector_stores=FakeVectorStores( + hits={ + "vs_fees": [ + search_hit( + file_id="file-fees", + filename="fees.pdf", + score=0.9, + texts=("Tuition is 100 per term.",), + ), + search_hit(file_id="file-faq", filename="faq.pdf", score=0.4), + ] + } + ) + ) + + +async def _run( + chat_model: ScriptedChatModel, + monkeypatch: pytest.MonkeyPatch, + *, + client: FakeOpenAIClient | None = None, + user_message: str | None = "What is the tuition?", + history: list[dict[str, str]] | None = None, + model_settings: ChatbotModelSettings | None = None, + collected_fields: dict[str, str] | None = None, +): + patch_chat_model(monkeypatch, chat_model) + return await run_rag_turn( + system_prompt=GOAL, + history=history or [], + user_message=user_message, + model_settings=model_settings or _model_settings(), + openai_client=client or _fees_client(), + collected_fields=collected_fields, + ) + + +class TestRunRagTurn: + @pytest.mark.asyncio + async def test_answers_without_searching( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + client = _fees_client() + chat_model = ScriptedChatModel(replies=["Hello! Ask me about fees."]) + + result = await _run( + chat_model, + monkeypatch, + client=client, + user_message="hi", + collected_fields={"name": "Asha"}, + ) + + assert result.reply == "Hello! Ask me about fees." + assert result.sources == [] + assert result.is_complete is False + assert result.collected_fields == {"name": "Asha"} + assert client.vector_stores.calls == [] + assert chat_model.bound_tool_names[0] == ["search_knowledge_base"] + + @pytest.mark.asyncio + async def test_tool_round_feeds_passages_back_and_returns_deduped_sources( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + client = _fees_client() + chat_model = ScriptedChatModel( + replies=[ + tool_call_reply("tuition fee", "fee payment deadline"), + "Tuition is 100 per term.", + ] + ) + + result = await _run(chat_model, monkeypatch, client=client) + + assert result.reply == "Tuition is 100 per term." + # Both searches hit the same two files; each file is reported once. + assert result.sources == [ + {"file_id": "file-fees", "filename": "fees.pdf"}, + {"file_id": "file-faq", "filename": "faq.pdf"}, + ] + assert [call["query"] for call in client.vector_stores.calls] == [ + "tuition fee", + "fee payment deadline", + ] + tool_results = [m for m in chat_model.calls[1] if isinstance(m, ToolMessage)] + assert len(tool_results) == 2 + assert ( + '\nTuition is 100 per term.\n' + in tool_results[0].content + ) + + @pytest.mark.asyncio + async def test_fourth_agent_call_is_forced_to_answer_after_three_tool_rounds( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + client = _fees_client() + chat_model = ScriptedChatModel( + replies=[ + tool_call_reply("tuition"), + tool_call_reply("tuition 2024"), + tool_call_reply("tuition per term"), + "Tuition is 100 per term.", + ] + ) + + result = await _run(chat_model, monkeypatch, client=client) + + assert result.reply == "Tuition is 100 per term." + assert chat_model.call_tool_choices == [None, None, None, {"type": "none"}] + assert len(client.vector_stores.calls) == 3 + + @pytest.mark.asyncio + async def test_invalid_tool_args_are_handed_back_to_the_model( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + client = _fees_client() + chat_model = ScriptedChatModel( + replies=[tool_call_reply(""), "Could you rephrase that?"] + ) + + result = await _run(chat_model, monkeypatch, client=client) + + assert result.reply == "Could you rephrase that?" + assert result.sources == [] + assert client.vector_stores.calls == [] + [tool_result] = [m for m in chat_model.calls[1] if isinstance(m, ToolMessage)] + assert tool_result.status == "error" + + @pytest.mark.asyncio + async def test_search_failure_fails_the_turn_instead_of_reaching_the_model( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + client = FakeOpenAIClient( + vector_stores=FakeVectorStores( + error=openai.APITimeoutError(request=_OPENAI_REQUEST) + ) + ) + chat_model = ScriptedChatModel( + replies=[tool_call_reply("tuition"), "should never be used"] + ) + + with pytest.raises(ChatbotProviderError, match="search timed out"): + await _run(chat_model, monkeypatch, client=client) + + assert len(chat_model.calls) == 1 + + @pytest.mark.asyncio + async def test_empty_final_reply_is_a_provider_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + with pytest.raises(ChatbotProviderError, match="EmptyReply"): + await _run(ScriptedChatModel(replies=[" "]), monkeypatch) + + @pytest.mark.asyncio + async def test_missing_message_with_history_rejected( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel() + + with pytest.raises(ChatbotInputError, match="`message` is required"): + await _run( + chat_model, + monkeypatch, + user_message=None, + history=[{"role": "assistant", "content": "Hi!"}], + ) + assert chat_model.calls == [] + + +class TestRagSystemPrompt: + @pytest.mark.asyncio + async def test_config_instructions_and_goal_follow_the_rag_instruction( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel(replies=["Namaste!"]) + + await _run( + chat_model, + monkeypatch, + model_settings=_model_settings(instructions="Answer in Hindi."), + ) + + system_message = chat_model.calls[0][0] + assert isinstance(system_message, SystemMessage) + assert system_message.content == ( + f"{RAG_INSTRUCTION}" + "\n\n\nAnswer in Hindi.\n" + f"\n\n\n{GOAL}\n" + ) + + @pytest.mark.asyncio + async def test_without_config_instructions_only_goal_is_added( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel(replies=["Hello!"]) + + await _run(chat_model, monkeypatch) + + system_message = chat_model.calls[0][0] + assert system_message.content == f"{RAG_INSTRUCTION}\n\n\n{GOAL}\n" + + +class TestRagHistorySummarization: + @pytest.mark.asyncio + async def test_history_over_budget_is_summarized_before_the_tool_loop( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + history: list[dict[str, str]] = [] + for i in range(4): + history.append({"role": "user", "content": f"user {i}"}) + history.append({"role": "assistant", "content": f"bot {i}"}) + chat_model = ScriptedChatModel( + replies=[SUMMARY_TEXT, tool_call_reply("tuition"), "Tuition is 100."], + token_count=5000, + ) + + result = await _run( + chat_model, + monkeypatch, + history=history, + model_settings=_model_settings(history_token_limit=4000), + ) + + assert result.reply == "Tuition is 100." + # One summary call, then two agent calls; the tool round is never summarized. + _summary_call, first_agent_call, second_agent_call = chat_model.calls + assert first_agent_call[1].content == ( + f"[Earlier conversation summary] {SUMMARY_TEXT}" + ) + assert isinstance(first_agent_call[-1], HumanMessage) + assert first_agent_call[-1].content == "What is the tuition?" + assert isinstance(second_agent_call[-1], ToolMessage) + assert "user 0" not in [m.content for m in second_agent_call] + + +class TestRagProviderErrorMapping: + @pytest.mark.parametrize( + ("error", "expected_message"), + [ + ( + anthropic.APITimeoutError(request=_ANTHROPIC_REQUEST), + "(code: APITimeoutError)", + ), + ( + anthropic.APIConnectionError(request=_ANTHROPIC_REQUEST), + "(code: APIConnectionError)", + ), + ( + ValueError("thinking.type=disabled is not supported"), + "[KAAPI] Unsupported model configuration for 'claude-sonnet-4-6': " + "thinking.type=disabled is not supported", + ), + ( + anthropic.APIStatusError( + "Overloaded", + response=httpx.Response( + 529, + request=_ANTHROPIC_REQUEST, + headers={"request-id": "req_123"}, + ), + body=None, + ), + "[ANTHROPIC] Chat model call failed: Overloaded " + "(code: 529, request_id: req_123)", + ), + ], + ids=["timeout", "connection", "value_error", "status"], + ) + @pytest.mark.asyncio + async def test_anthropic_errors_become_provider_errors( + self, + monkeypatch: pytest.MonkeyPatch, + error: Exception, + expected_message: str, + ) -> None: + with pytest.raises(ChatbotProviderError) as exc_info: + await _run(ScriptedChatModel(error=error), monkeypatch) + + assert expected_message in str(exc_info.value) + assert exc_info.value.__cause__ is error diff --git a/backend/app/tests/services/chatbot/agents/rag/test_retriever.py b/backend/app/tests/services/chatbot/agents/rag/test_retriever.py new file mode 100644 index 000000000..2a1b6a05b --- /dev/null +++ b/backend/app/tests/services/chatbot/agents/rag/test_retriever.py @@ -0,0 +1,243 @@ +import httpx +import openai +import pytest +from langchain_core.messages import ToolMessage +from sqlmodel import Session + +from app.services.chatbot.agents.rag.prompts import NO_PASSAGES_FOUND +from app.services.chatbot.agents.rag.retriever import ( + build_search_tool, + format_passages, + get_knowledge_base_client, + search_knowledge_bases, +) +from app.services.chatbot.agents.rag.types import KnowledgeBasePassage +from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError +from app.tests.utils.chatbot import FakeOpenAIClient, FakeVectorStores, search_hit +from app.tests.utils.test_data import create_test_credential, create_test_project + +_OPENAI_REQUEST = httpx.Request("POST", "https://api.openai.com/v1/vector_stores") + + +def _client(**hits: list) -> FakeOpenAIClient: + return FakeOpenAIClient(vector_stores=FakeVectorStores(hits=hits)) + + +def _hits(prefix: str, count: int) -> list: + return [ + search_hit(file_id=f"{prefix}-{i}", filename=f"{prefix}-{i}.pdf", score=i / 100) + for i in range(count) + ] + + +class TestSearchKnowledgeBases: + def test_merges_knowledge_bases_and_keeps_overall_top_by_score(self) -> None: + client = _client( + vs_a=[ + search_hit(file_id="a-low", filename="a.pdf", score=0.2), + search_hit(file_id="a-high", filename="a.pdf", score=0.9), + ], + vs_b=[ + search_hit(file_id="b-mid", filename="b.pdf", score=0.5), + search_hit(file_id="b-high", filename="b.pdf", score=0.7), + ], + ) + + passages = search_knowledge_bases( + client=client, + knowledge_base_ids=("vs_a", "vs_b"), + query="fees", + max_num_results=3, + ) + + assert [p.file_id for p in passages] == ["a-high", "b-high", "b-mid"] + assert client.vector_stores.calls == [ + {"vector_store_id": "vs_a", "query": "fees", "max_num_results": 3}, + {"vector_store_id": "vs_b", "query": "fees", "max_num_results": 3}, + ] + + def test_defaults_to_five_results(self) -> None: + client = _client(vs_a=_hits("a", 7)) + + passages = search_knowledge_bases( + client=client, knowledge_base_ids=("vs_a",), query="q", max_num_results=None + ) + + assert len(passages) == 5 + assert client.vector_stores.calls[0]["max_num_results"] == 5 + + def test_caps_requested_results_at_fifty(self) -> None: + client = _client(vs_a=_hits("a", 60)) + + passages = search_knowledge_bases( + client=client, knowledge_base_ids=("vs_a",), query="q", max_num_results=80 + ) + + assert len(passages) == 50 + assert client.vector_stores.calls[0]["max_num_results"] == 50 + + def test_chunks_of_a_hit_are_joined_into_one_passage(self) -> None: + client = _client( + vs_a=[ + search_hit( + file_id="f1", + filename="guide.pdf", + score=0.8, + texts=("first chunk", "second chunk"), + ) + ] + ) + + passages = search_knowledge_bases( + client=client, knowledge_base_ids=("vs_a",), query="q", max_num_results=None + ) + + assert passages == [ + KnowledgeBasePassage( + file_id="f1", + filename="guide.pdf", + score=0.8, + text="first chunk\nsecond chunk", + ) + ] + + @pytest.mark.parametrize( + ("error", "expected_message"), + [ + ( + openai.APITimeoutError(request=_OPENAI_REQUEST), + "[KAAPI] Knowledge base search timed out — retry the turn. If " + "persistent, contact Kaapi. (code: APITimeoutError)", + ), + ( + openai.APIConnectionError(request=_OPENAI_REQUEST), + "[KAAPI] Network/DNS issue reaching OpenAI for knowledge base search — " + "check connectivity. If persistent, contact Kaapi. " + "(code: APIConnectionError)", + ), + ( + openai.NotFoundError( + "No vector store found", + response=httpx.Response( + 404, + request=_OPENAI_REQUEST, + headers={"x-request-id": "req_123"}, + ), + body=None, + ), + "[OPENAI] Knowledge base search failed for 'vs_a': No vector store " + "found (code: 404, request_id: req_123)", + ), + ( + openai.OpenAIError("unexpected payload"), + "[OPENAI] Knowledge base search failed for 'vs_a': unexpected payload " + "— retry the turn. If persistent, contact Kaapi. (code: OpenAIError)", + ), + ], + ids=["timeout", "connection", "status", "other"], + ) + def test_openai_errors_become_provider_errors( + self, error: openai.OpenAIError, expected_message: str + ) -> None: + client = FakeOpenAIClient(vector_stores=FakeVectorStores(error=error)) + + with pytest.raises(ChatbotProviderError) as exc_info: + search_knowledge_bases( + client=client, + knowledge_base_ids=("vs_a",), + query="q", + max_num_results=None, + ) + + assert str(exc_info.value) == expected_message + assert exc_info.value.__cause__ is error + + +class TestFormatPassages: + def test_no_passages_tells_the_model_nothing_was_found(self) -> None: + assert format_passages([]) == NO_PASSAGES_FOUND + + def test_passages_are_wrapped_with_escaped_filename(self) -> None: + passages = [ + KnowledgeBasePassage( + file_id="f1", filename='fees "2024".pdf', score=0.9, text="Fee is 100." + ), + KnowledgeBasePassage( + file_id="f2", filename="a&b.pdf", score=0.5, text="Due in May." + ), + ] + + assert format_passages(passages) == ( + '\nFee is 100.\n' + "\n\n" + '\nDue in May.\n' + ) + + +class TestBuildSearchTool: + def test_tool_returns_passages_as_content_and_hits_as_artifact(self) -> None: + client = _client( + vs_a=[search_hit(file_id="f1", filename="fees.pdf", score=0.9)] + ) + tool = build_search_tool( + client=client, knowledge_base_ids=("vs_a",), max_num_results=None + ) + + message = tool.invoke( + { + "type": "tool_call", + "name": "search_knowledge_base", + "args": {"query": "fees"}, + "id": "toolu_1", + } + ) + + assert isinstance(message, ToolMessage) + assert message.content == '\ntext\n' + assert message.artifact == [ + KnowledgeBasePassage( + file_id="f1", filename="fees.pdf", score=0.9, text="text" + ) + ] + assert client.vector_stores.calls[0]["query"] == "fees" + + def test_no_hits_returns_not_found_text_and_empty_artifact(self) -> None: + tool = build_search_tool( + client=_client(), knowledge_base_ids=("vs_a",), max_num_results=None + ) + + message = tool.invoke( + { + "type": "tool_call", + "name": "search_knowledge_base", + "args": {"query": "fees"}, + "id": "toolu_1", + } + ) + + assert message.content == NO_PASSAGES_FOUND + assert message.artifact == [] + + +class TestGetKnowledgeBaseClient: + def test_project_without_openai_credentials_rejected(self, db: Session) -> None: + project = create_test_project(db) + + with pytest.raises(ChatbotInputError, match="OpenAI credentials"): + get_knowledge_base_client( + session=db, + organization_id=project.organization_id, + project_id=project.id, + ) + + def test_client_uses_project_api_key(self, db: Session) -> None: + _credentials, project = create_test_credential(db) + + client = get_knowledge_base_client( + session=db, + organization_id=project.organization_id, + project_id=project.id, + ) + + assert client.api_key.startswith("sk-") + assert client.max_retries == 1 diff --git a/backend/app/tests/services/chatbot/agents/test_registry.py b/backend/app/tests/services/chatbot/agents/test_registry.py new file mode 100644 index 000000000..31a9a24f7 --- /dev/null +++ b/backend/app/tests/services/chatbot/agents/test_registry.py @@ -0,0 +1,92 @@ +import pytest + +from app.models import ChatbotFlowNodeTypeEnum +from app.services.chatbot.agents.registry import run_agent_turn +from app.services.chatbot.exceptions import ChatbotInputError +from app.services.chatbot.utils import ChatbotModelSettings +from app.tests.utils.chatbot import ( + FakeOpenAIClient, + FakeVectorStores, + ScriptedChatModel, + patch_chat_model, + search_hit, + tool_call_reply, +) + +SYSTEM_PROMPT = "Collect the user's name and city." +_MODEL_SETTINGS = ChatbotModelSettings( + model="claude-sonnet-4-6", knowledge_base_ids=("vs_fees",) +) + + +class TestRunAgentTurn: + @pytest.mark.asyncio + async def test_rag_without_openai_client_rejected_before_calling_model( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel(replies=["should never be used"]) + patch_chat_model(monkeypatch, chat_model) + + with pytest.raises(ChatbotInputError, match="needs an OpenAI client"): + await run_agent_turn( + node_type=ChatbotFlowNodeTypeEnum.RAG, + system_prompt=SYSTEM_PROMPT, + history=[], + user_message="hello", + model_settings=_MODEL_SETTINGS, + ) + assert chat_model.calls == [] + + @pytest.mark.asyncio + async def test_onboarding_runs_field_extraction_and_ignores_client( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel(replies=["Which city?"]) + patch_chat_model(monkeypatch, chat_model) + client = FakeOpenAIClient() + + result = await run_agent_turn( + node_type=ChatbotFlowNodeTypeEnum.ONBOARDING, + system_prompt=SYSTEM_PROMPT, + history=[], + user_message="I'm Asha", + model_settings=_MODEL_SETTINGS, + openai_client=client, + ) + + assert result.reply == "Which city?" + # Only the onboarding agent runs the structured-output extraction step. + assert len(chat_model.extraction_calls) == 1 + assert chat_model.bound_tool_names == [] + assert client.vector_stores.calls == [] + + @pytest.mark.asyncio + async def test_rag_searches_with_the_given_client( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + chat_model = ScriptedChatModel( + replies=[tool_call_reply("tuition"), "Tuition is 100."] + ) + patch_chat_model(monkeypatch, chat_model) + client = FakeOpenAIClient( + vector_stores=FakeVectorStores( + hits={ + "vs_fees": [ + search_hit(file_id="file-fees", filename="fees.pdf", score=0.9) + ] + } + ) + ) + + result = await run_agent_turn( + node_type=ChatbotFlowNodeTypeEnum.RAG, + system_prompt=SYSTEM_PROMPT, + history=[], + user_message="What is the tuition?", + model_settings=_MODEL_SETTINGS, + openai_client=client, + ) + + assert result.reply == "Tuition is 100." + assert result.sources == [{"file_id": "file-fees", "filename": "fees.pdf"}] + assert chat_model.extraction_calls == [] diff --git a/backend/app/tests/services/chatbot/test_config.py b/backend/app/tests/services/chatbot/test_config.py index d02fb4fc4..42f997696 100644 --- a/backend/app/tests/services/chatbot/test_config.py +++ b/backend/app/tests/services/chatbot/test_config.py @@ -5,10 +5,12 @@ from fastapi import HTTPException from sqlmodel import Session +from app.crud import CollectionCrud from app.crud.config import ConfigVersionCrud -from app.models import Config, Project +from app.models import Config, Project, ProviderType from app.services.chatbot.config import ( get_anthropic_api_key, + prepare_rag_search_client, resolve_chatbot_model_settings, ) from app.services.chatbot.exceptions import ChatbotInputError @@ -18,7 +20,13 @@ create_test_anthropic_config, create_test_anthropic_credential, ) -from app.tests.utils.test_data import create_test_config, create_test_project +from app.tests.utils.collection import get_vector_store_collection +from app.tests.utils.test_data import ( + build_anthropic_text_config_blob, + create_test_config, + create_test_credential, + create_test_project, +) MISSING_CREDENTIALS = "Anthropic credentials are not configured for this project" @@ -108,6 +116,33 @@ def test_latest_version_with_server_defaults(self, db: Session) -> None: history_token_limit=4000, ) + def test_rag_params_are_carried_into_settings(self, db: Session) -> None: + config = create_test_config( + db, + config_blob=build_anthropic_text_config_blob( + instructions="Answer in Hindi.", + knowledge_base_ids=["vs_alpha", "vs_beta"], + max_num_results=8, + ), + ) + create_test_anthropic_credential(db, project_id=config.project_id) + + model_settings = _resolve(db, config) + + assert model_settings.instructions == "Answer in Hindi." + assert model_settings.knowledge_base_ids == ("vs_alpha", "vs_beta") + assert model_settings.max_num_results == 8 + + def test_absent_rag_params_leave_empty_defaults(self, db: Session) -> None: + config = create_test_config(db, config_blob=build_anthropic_text_config_blob()) + create_test_anthropic_credential(db, project_id=config.project_id) + + model_settings = _resolve(db, config) + + assert model_settings.instructions is None + assert model_settings.knowledge_base_ids == () + assert model_settings.max_num_results is None + def test_settings_carry_the_projects_anthropic_key(self, db: Session) -> None: config = create_test_anthropic_config(db, with_anthropic_credentials=False) create_test_anthropic_credential( @@ -255,3 +290,95 @@ def test_non_anthropic_effort_rejected(self, db: Session, effort: str) -> None: with pytest.raises(ChatbotInputError, match=f"'{effort}' is not an Anthropic"): _resolve(db, config) + + +def _prepare(project: Project, *knowledge_base_ids: str, db: Session): + return prepare_rag_search_client( + session=db, + organization_id=project.organization_id, + project_id=project.id, + model_settings=ChatbotModelSettings( + model="claude-sonnet-4-6", knowledge_base_ids=knowledge_base_ids + ), + ) + + +@pytest.fixture +def project_with_openai_creds(db: Session) -> Project: + _credentials, project = create_test_credential(db) + return project + + +class TestPrepareRagSearchClient: + def test_known_collections_and_credentials_return_project_client( + self, db: Session, project_with_openai_creds: Project + ) -> None: + first = get_vector_store_collection(db, project_with_openai_creds) + second = get_vector_store_collection(db, project_with_openai_creds) + + client = _prepare( + project_with_openai_creds, + first.knowledge_base_id, + second.knowledge_base_id, + db=db, + ) + + assert client.api_key.startswith("sk-") + + def test_empty_knowledge_base_ids_rejected( + self, db: Session, project_with_openai_creds: Project + ) -> None: + with pytest.raises(ChatbotInputError, match="non-empty"): + _prepare(project_with_openai_creds, db=db) + + def test_unknown_id_is_named_and_known_id_is_not( + self, db: Session, project_with_openai_creds: Project + ) -> None: + known = get_vector_store_collection(db, project_with_openai_creds) + + with pytest.raises(ChatbotInputError) as exc_info: + _prepare( + project_with_openai_creds, + known.knowledge_base_id, + "vs_missing", + db=db, + ) + + assert str(exc_info.value).endswith( + "collections: vs_missing; use the knowledge_base_id of an existing collection." + ) + assert known.knowledge_base_id not in str(exc_info.value) + + def test_deleted_collection_rejected( + self, db: Session, project_with_openai_creds: Project + ) -> None: + collection = get_vector_store_collection(db, project_with_openai_creds) + CollectionCrud(db, project_with_openai_creds.id).delete_by_id(collection.id) + + with pytest.raises(ChatbotInputError, match=collection.knowledge_base_id): + _prepare(project_with_openai_creds, collection.knowledge_base_id, db=db) + + def test_non_openai_collection_rejected( + self, db: Session, project_with_openai_creds: Project + ) -> None: + collection = get_vector_store_collection( + db, project_with_openai_creds, provider=ProviderType.google_aistudio + ) + + with pytest.raises(ChatbotInputError, match=collection.knowledge_base_id): + _prepare(project_with_openai_creds, collection.knowledge_base_id, db=db) + + def test_other_projects_collection_rejected( + self, db: Session, project_with_openai_creds: Project + ) -> None: + foreign = get_vector_store_collection(db, create_test_project(db)) + + with pytest.raises(ChatbotInputError, match=foreign.knowledge_base_id): + _prepare(project_with_openai_creds, foreign.knowledge_base_id, db=db) + + def test_missing_openai_credentials_rejected(self, db: Session) -> None: + project = create_test_project(db) + collection = get_vector_store_collection(db, project) + + with pytest.raises(ChatbotInputError, match="OpenAI credentials"): + _prepare(project, collection.knowledge_base_id, db=db) diff --git a/backend/app/tests/services/chatbot/test_session.py b/backend/app/tests/services/chatbot/test_session.py index 295a8a8fc..e8156f605 100644 --- a/backend/app/tests/services/chatbot/test_session.py +++ b/backend/app/tests/services/chatbot/test_session.py @@ -4,9 +4,12 @@ from uuid import uuid4 import anthropic +import httpx +import openai import pytest from sqlmodel import Session +from app.crud import CollectionCrud from app.crud.chatbot import ( list_chatbot_session_messages, update_chatbot_session_progress, @@ -15,11 +18,13 @@ from app.models import ( Chatbot, ChatbotCompletionStatusEnum, + ChatbotFlowNodeTypeEnum, ChatbotMessageRoleEnum, ChatbotSession, Config, Project, ) +from app.services.chatbot import config as config_service from app.services.chatbot import session as session_service from app.services.chatbot.agents.onboarding import graph as onboarding_graph from app.services.chatbot.agents.onboarding.types import FieldExtraction @@ -31,6 +36,8 @@ from app.tests.utils.chatbot import ( ANTHROPIC_REQUEST, DEFAULT_SYSTEM_PROMPT, + FakeOpenAIClient, + FakeVectorStores, ONBOARDING_NODE_ID, WEBHOOK_URL, ScriptedChatModel, @@ -44,8 +51,15 @@ create_test_live_chatbot, extracted_fields, patch_chat_model, + search_hit, + tool_call_reply, +) +from app.tests.utils.collection import get_vector_store_collection +from app.tests.utils.test_data import ( + create_test_config, + create_test_credential, + create_test_project, ) -from app.tests.utils.test_data import create_test_config ChatTurns = list[tuple[ChatbotMessageRoleEnum, str]] @@ -140,17 +154,20 @@ def _success_webhook( reply: str, completion_status: str = "in_progress", collected_fields: dict[str, str] | None = None, + node_id: str = ONBOARDING_NODE_ID, + sources: list[Any] | None = None, ) -> 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, "collected_fields": collected_fields or {}, + "sources": sources or [], }, } @@ -705,7 +722,7 @@ def test_active_node_missing_from_flow_posts_failure_webhook( ) ] - def test_non_onboarding_active_node_posts_failure_webhook( + def test_unsupported_active_node_type_posts_failure_webhook( self, db: Session, config: Config, @@ -725,7 +742,8 @@ def test_non_onboarding_active_node_posts_failure_webhook( [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"] + # onboarding and rag both run now; only the others are rejected. + assert "only onboarding, rag nodes can run" in payload["error"] @pytest.mark.parametrize( "node_overrides", @@ -754,3 +772,157 @@ def test_node_without_prompt_or_config_posts_failure_webhook( _assert_single_failure_webhook( callbacks, chatbot_session, "missing data.prompt or data.config_id" ) + + +RAG_GOAL = "Answer questions about school fees from the documents." +RAG_NODE_ID = "rag-1" +_OPENAI_REQUEST = httpx.Request("POST", "https://api.openai.com/v1/vector_stores") + + +@pytest.fixture +def vector_stores(monkeypatch: pytest.MonkeyPatch) -> FakeVectorStores: + stores = FakeVectorStores() + real_get_client = config_service.get_knowledge_base_client + + def _get_client(**kwargs: Any) -> FakeOpenAIClient: + # The real lookup still enforces the project's OpenAI credentials; only the + # network-facing client is swapped. + real_get_client(**kwargs) + return FakeOpenAIClient(vector_stores=stores) + + monkeypatch.setattr(config_service, "get_knowledge_base_client", _get_client) + return stores + + +def _rag_session(db: Session, project: Project) -> tuple[ChatbotSession, str]: + """A session whose live flow's entry node is a rag node over one collection.""" + collection = get_vector_store_collection(db, project) + rag_config = create_test_anthropic_config( + db, + project_id=project.id, + knowledge_base_ids=[collection.knowledge_base_id], + ) + rag_node = build_onboarding_node( + config_id=rag_config.id, + prompt=RAG_GOAL, + node_id=RAG_NODE_ID, + node_type=ChatbotFlowNodeTypeEnum.RAG.value, + ) + chatbot_session = _session_on_live_flow( + db, rag_config, nodes=[rag_node], active_node_id=RAG_NODE_ID + ) + add_test_chatbot_messages( + db, chatbot_session, [(ChatbotMessageRoleEnum.USER, "What is the tuition?")] + ) + return chatbot_session, collection.knowledge_base_id + + +class TestRagSessionTurn: + def test_answer_with_sources_is_stored_and_posted( + self, + db: Session, + callbacks: CallbackRecorder, + vector_stores: FakeVectorStores, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _credentials, project = create_test_credential(db) + chatbot_session, knowledge_base_id = _rag_session(db, project) + vector_stores.hits[knowledge_base_id] = [ + search_hit( + file_id="file-fees", + filename="fees.pdf", + score=0.9, + texts=("Tuition is 100 per term.",), + ) + ] + chat_model = ScriptedChatModel( + replies=[tool_call_reply("tuition fee"), "Tuition is 100 per term."] + ) + patch_chat_model(monkeypatch, chat_model) + + chatbot = db.get(Chatbot, chatbot_session.chatbot_id) + assert chatbot is not None + _process(chatbot_session, db) + + assert callbacks.payloads == [ + _success_webhook( + chatbot_session, + chatbot, + reply="Tuition is 100 per term.", + node_id=RAG_NODE_ID, + sources=[{"file_id": "file-fees", "filename": "fees.pdf"}], + ) + ] + assert _transcript(db, chatbot_session) == [ + ("user", "What is the tuition?"), + ("assistant", "Tuition is 100 per term."), + ] + assert vector_stores.calls[0]["vector_store_id"] == knowledge_base_id + + def test_missing_openai_credentials_post_failure_without_calling_model( + self, + db: Session, + callbacks: CallbackRecorder, + vector_stores: FakeVectorStores, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + chatbot_session, _kb_id = _rag_session(db, create_test_project(db)) + chat_model = ScriptedChatModel(replies=["should never be used"]) + patch_chat_model(monkeypatch, chat_model) + + _process(chatbot_session, db) + + assert chat_model.calls == [] + assert vector_stores.calls == [] + [payload] = callbacks.payloads + assert payload["success"] is False + assert "OpenAI credentials" in payload["error"] + assert payload["data"] == {"session_id": str(chatbot_session.id)} + assert _transcript(db, chatbot_session) == [("user", "What is the tuition?")] + + def test_knowledge_base_deleted_after_create_posts_failure( + self, + db: Session, + callbacks: CallbackRecorder, + vector_stores: FakeVectorStores, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _credentials, project = create_test_credential(db) + chatbot_session, knowledge_base_id = _rag_session(db, project) + crud = CollectionCrud(db, project.id) + [collection] = crud.read_all() + crud.delete_by_id(collection.id) + chat_model = ScriptedChatModel(replies=["should never be used"]) + patch_chat_model(monkeypatch, chat_model) + + _process(chatbot_session, db) + + assert chat_model.calls == [] + [payload] = callbacks.payloads + assert payload["success"] is False + assert knowledge_base_id in payload["error"] + + def test_search_failure_posts_failure_and_stores_no_reply( + self, + db: Session, + callbacks: CallbackRecorder, + vector_stores: FakeVectorStores, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _credentials, project = create_test_credential(db) + chatbot_session, _kb_id = _rag_session(db, project) + vector_stores.error = openai.APIConnectionError(request=_OPENAI_REQUEST) + chat_model = ScriptedChatModel( + replies=[tool_call_reply("tuition fee"), "should never be used"] + ) + patch_chat_model(monkeypatch, chat_model) + + _process(chatbot_session, db) + + [payload] = callbacks.payloads + assert payload["success"] is False + assert "(code: APIConnectionError)" in payload["error"] + assert "knowledge base search" in payload["error"] + assert payload["data"] == {"session_id": str(chatbot_session.id)} + assert _transcript(db, chatbot_session) == [("user", "What is the tuition?")] + assert len(chat_model.calls) == 1 diff --git a/backend/app/tests/utils/chatbot.py b/backend/app/tests/utils/chatbot.py index 7a9b68ba2..5b1b27295 100644 --- a/backend/app/tests/utils/chatbot.py +++ b/backend/app/tests/utils/chatbot.py @@ -1,4 +1,5 @@ from collections.abc import Sequence +from dataclasses import dataclass, field from typing import Any from uuid import UUID @@ -8,6 +9,8 @@ from langchain_core.messages import AIMessage, BaseMessage from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.runnables import Runnable, RunnableLambda +from openai.types import VectorStoreSearchResponse +from openai.types.vector_store_search_response import Content from pydantic import Field from sqlmodel import Session, select @@ -36,6 +39,7 @@ ) from app.models.provider_credentials import Provider from app.services.chatbot.agents.onboarding import graph as onboarding_graph +from app.services.chatbot.agents.rag import graph as rag_graph from app.services.chatbot.agents.onboarding.types import ( ExtractedField, FieldExtraction, @@ -98,7 +102,8 @@ def extracted_fields(fields: dict[str, str]) -> list[ExtractedField]: class ScriptedChatModel(BaseChatModel): """Stands in for ChatAnthropic: replies from a script and records each call.""" - replies: list[str] = Field(default_factory=list) + # A str becomes a plain AIMessage; pass an AIMessage to script tool calls. + replies: list[str | AIMessage] = Field(default_factory=list) token_count: int = 0 error: Exception | None = None calls: list[list[BaseMessage]] = Field(default_factory=list) @@ -107,6 +112,9 @@ 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) + # tool_choice each call was bound with (None = unbound or default "auto"). + call_tool_choices: list[Any] = Field(default_factory=list) + bound_tool_names: list[list[str]] = Field(default_factory=list) @property def _llm_type(self) -> str: @@ -120,11 +128,19 @@ def _generate( **kwargs: Any, ) -> ChatResult: self.calls.append(list(messages)) + self.call_tool_choices.append(kwargs.get("tool_choice")) if self.error is not None: raise self.error - reply = AIMessage(content=self.replies.pop(0)) + scripted = self.replies.pop(0) + reply = scripted if isinstance(scripted, AIMessage) else AIMessage(scripted) return ChatResult(generations=[ChatGeneration(message=reply)]) + def bind_tools( + self, tools: Sequence[Any], *, tool_choice: Any = None, **kwargs: Any + ) -> Runnable[Any, BaseMessage]: + self.bound_tool_names.append([tool.name for tool in tools]) + return self.bind(tool_choice=tool_choice) + def get_num_tokens_from_messages( self, messages: list[BaseMessage], tools: Sequence[Any] | None = None ) -> int: @@ -149,12 +165,28 @@ 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 graph_module in (onboarding_graph, rag_graph): + monkeypatch.setattr( + graph_module, "get_chat_model", lambda _settings: chat_model + ) return chat_model +def tool_call_reply(*queries: str, text: str = "") -> AIMessage: + """An agent reply that calls the knowledge-base search once per query.""" + return AIMessage( + content=text, + tool_calls=[ + { + "name": "search_knowledge_base", + "args": {"query": query}, + "id": f"toolu_{index}_{query.replace(' ', '_')}", + } + for index, query in enumerate(queries) + ], + ) + + def create_test_anthropic_credential( db: Session, *, project_id: int, api_key: str | None = None ) -> str: @@ -309,6 +341,51 @@ def create_test_live_chatbot( return chatbot, flow_version +def search_hit( + *, file_id: str, filename: str, score: float, texts: Sequence[str] = ("text",) +) -> VectorStoreSearchResponse: + return VectorStoreSearchResponse( + attributes=None, + content=[Content(text=text, type="text") for text in texts], + file_id=file_id, + filename=filename, + score=score, + ) + + +@dataclass +class _SearchPage: + data: list[VectorStoreSearchResponse] + + +@dataclass +class FakeVectorStores: + hits: dict[str, list[VectorStoreSearchResponse]] = field(default_factory=dict) + error: Exception | None = None + calls: list[dict[str, Any]] = field(default_factory=list) + + def search( + self, vector_store_id: str, *, query: str, max_num_results: int + ) -> _SearchPage: + self.calls.append( + { + "vector_store_id": vector_store_id, + "query": query, + "max_num_results": max_num_results, + } + ) + if self.error is not None: + raise self.error + return _SearchPage(data=list(self.hits.get(vector_store_id, []))) + + +@dataclass +class FakeOpenAIClient: + """Stands in for openai.OpenAI; only vector-store search is used by the rag agent.""" + + vector_stores: FakeVectorStores = field(default_factory=FakeVectorStores) + + def create_test_chatbot_session( db: Session, *, diff --git a/backend/app/tests/utils/collection.py b/backend/app/tests/utils/collection.py index d662f339d..212b11ce5 100644 --- a/backend/app/tests/utils/collection.py +++ b/backend/app/tests/utils/collection.py @@ -31,6 +31,7 @@ def get_vector_store_collection( *, vector_store_id: Optional[str] = None, collection_id: Optional[UUID] = None, + provider: ProviderType = ProviderType.openai, ) -> Collection: """ Create a Collection configured for the Vector Store path. @@ -42,9 +43,9 @@ def get_vector_store_collection( collection = Collection( id=collection_id or uuid4(), project_id=project.id, - knowledge_base_provider=get_service_name("openai"), + knowledge_base_provider=get_service_name(provider.value), knowledge_base_id=vector_store_id, - provider=ProviderType.openai, + provider=provider, ) return CollectionCrud(db, project.id).create(collection) diff --git a/docs/wiki/domain-map.md b/docs/wiki/domain-map.md index 37759e939..2a8dbfe34 100644 --- a/docs/wiki/domain-map.md +++ b/docs/wiki/domain-map.md @@ -37,7 +37,7 @@ APIKey → Organization, Project, User # programmatic access | TTSResult | tts_evaluation.py | Run, Org, Project | human annotation UI (logical) | | Assessment / AssessmentRun | assessment.py | Config, AssessmentSubmission, BatchJob, Job, Org, Project | console UI (logical); webhook consumers (logical) | | Document | document.py | Project, Document (parent) | DocumentCollection, DocTransformationJob, FineTuning, ModelEvaluation | -| Collection | collection.py | Project | DocumentCollection, CollectionJob; provider vector stores (logical) | +| Collection | collection.py | Project | DocumentCollection, CollectionJob; provider vector stores (logical); rag chatbot turns search OpenAI collections by `knowledge_base_id` from the config (logical) | | DocumentCollection | document_collection.py | Document, Collection | RAG lookups (logical) | | CollectionJob | collection_job.py | Collection, Project | async collection create/delete (logical) | | DocTransformationJob | doc_transformation_job.py | Document | doctransform service (logical) | diff --git a/docs/wiki/modules/chatbot.md b/docs/wiki/modules/chatbot.md index c21bfabca..95e4cd4e5 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, using the **project's own Anthropic credentials** (no platform fallback). 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 two agent node types: `onboarding` ("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 `rag` (answers questions from the config's knowledge bases — OpenAI vector stores searched with the project's OpenAI credentials — over a tool-calling LangGraph; the node prompt is the goal and decides scope). `coordinator` nodes can be stored in a flow but are not runnable yet. No deep-dive doc yet. All paths relative to `backend/app/`. @@ -23,36 +23,50 @@ All paths relative to `backend/app/`. ## Tables (SQLModel, `models/chatbot.py`, migration `088`) - `chatbot` (`Chatbot`) — `name`, org/project FKs (CASCADE), `deleted_at` soft delete (sessions reference it with RESTRICT). - `chatbot_flow_version` (`ChatbotFlowVersion`) — `chatbot_id` (CASCADE), `version` (UNIQUE with chatbot_id, `uq_chatbot_flow_version_chatbot_version`), `nodes` / `edges` JSONB stored as the UI sent them (React Flow camelCase keys, extra UI keys round-trip), `is_live` with partial unique index `uq_chatbot_flow_version_live_chatbot` (`WHERE is_live`). The highest version is the draft when not live; live versions are immutable. -- `chatbot_session` (`ChatbotSession`) — `chatbot_id` (RESTRICT), `flow_version_id` (RESTRICT, pinned at create), `contact_ref`, `active_node_id` (React Flow node id inside the pinned flow), `status` (`ChatbotSessionStatusEnum` active/inactive), `completion_status` (`ChatbotCompletionStatusEnum` in_progress/complete/abandoned), `collected_fields` (JSON flat `{field_name: value}`). No org/project columns: tenant scoping joins through `chatbot`. Indexes: `(chatbot_id, contact_ref)`, `contact_ref` (cross-chatbot latest lookup), partial unique `uq_chatbot_session_active_chatbot_contact` on `(chatbot_id, contact_ref) WHERE status = 'active'`. +- `chatbot_session` (`ChatbotSession`) — `chatbot_id` (RESTRICT), `flow_version_id` (RESTRICT, pinned at create), `contact_ref`, `active_node_id` (React Flow node id inside the pinned flow, decides which agent runs each turn), `status` (`ChatbotSessionStatusEnum` active/inactive), `completion_status` (`ChatbotCompletionStatusEnum` in_progress/complete/abandoned), `collected_fields` (JSON flat `{field_name: value}`; `rag` sessions never populate it). No org/project columns, no agent-type column: tenant scoping joins through `chatbot`, and which agent runs is read from the pinned flow's active node type, not stored redundantly on the session. Indexes: `(chatbot_id, contact_ref)`, `contact_ref` (cross-chatbot latest lookup), partial unique `uq_chatbot_session_active_chatbot_contact` on `(chatbot_id, contact_ref) WHERE status = 'active'`. - `chatbot_session_message` (`ChatbotSessionMessage`) — ordered transcript rows (`role` user/assistant), `node_id` = the flow node that handled the turn. -- 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. +- Session lifecycle: one active session per contact + chatbot. `active`+`in_progress` → `active`+`complete` as fields are collected (onboarding only). 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` = coordinator/onboarding/rag) 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; a flow-level check that the project has Anthropic credentials if it has any agent node; `coordinator` rejected as not supported yet (`flow.py:SUPPORTED_AGENT_NODE_TYPES` = onboarding, rag); every `onboarding`/`rag` node has `config_id` + non-empty `prompt` and its config passes `get_chatbot_config_version` + `build_chatbot_model_settings`; a `rag` node additionally passes `prepare_rag_search_client` (non-empty, known `knowledge_base_ids` + OpenAI credentials); 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`) +## Turn flow (onboarding node, `services/chatbot/agents/onboarding/graph.py`) 1. `extract_fields_node` — skipped until there is a real user turn (opening greeting). Otherwise calls the model with structured output (`FieldExtraction{collected_fields, missing_fields, is_complete}`) over the node prompt, the previously collected map, and the (summarized) conversation. Uses Anthropic native structured output (`with_structured_output(..., method="json_schema")`) with the config's own thinking/effort, because forced tool use is rejected with thinking on and Opus 5.5 can't disable thinking. Merged in code: `{**prior, **extracted}`; `is_complete` only when nothing is missing. 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. +3. `run_chatbot_turn(..., collected_fields=prior)` → `ChatbotTurnResult{reply, collected_fields, is_complete, sources=[]}` (`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` via the shared `provider_errors.to_provider_error`, so the session path always sends a failure webhook. + +## Turn flow (rag node, `services/chatbot/agents/rag/graph.py`) +1. `prepare_context_node` — runs `summarize_if_needed` once and replaces the state messages (`RemoveMessage(REMOVE_ALL_MESSAGES)`), so the tool loop is never summarized mid tool_use/tool_result pair. +2. `agent_node` — chat model with `search_knowledge_base` bound (`tool_choice` auto; forced tool use is rejected with thinking). System prompt = `RAG_INSTRUCTION` + `` (config `params.instructions`, if set) + `` (the node's `prompt`). After `MAX_TOOL_ROUNDS` (3) tool-calling AIMessages it is re-bound with `tool_choice={"type": "none"}` so it must answer. +3. `tools_condition` → `ToolNode` (`tools`) → back to `agent_node`; END once the model stops calling tools. Only bad tool args (`ToolInvocationError`) are fed back to the model; a search failure (`ChatbotProviderError`) fails the turn. +4. `search_knowledge_base(query)` (`retriever.py`, `content_and_artifact`) — `client.vector_stores.search` on every `knowledge_base_ids` entry (from the node's config), merged, sorted by score, top `max_num_results` (default 5, capped at 50); content is `text` passages, artifact the `KnowledgeBasePassage` list. OpenAI errors → `ChatbotProviderError` (`[OPENAI] Knowledge base search failed ...`; timeouts/connection `[KAAPI]`). +5. `run_rag_turn(..., openai_client)` → `ChatbotTurnResult{reply, collected_fields (unchanged), is_complete=False, sources}`; `sources` are the retrieved files deduped by `file_id`. ## 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/agents/onboarding/{graph,prompts,types}.py` — `run_chatbot_turn`, the extraction/reply nodes, `FieldExtraction` -- `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/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`, `get_chatbot_or_raise` +- `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 via `_resolve_turn_node`; inactive session, missing node, or a node type the runtime doesn't support → failure webhook; for a `rag` node builds the OpenAI client in the first DB block (missing credentials / bad knowledge bases → failure webhook); dispatches to `agents/registry.py:run_agent_turn` with the stored `collected_fields`, stores the reply with `node_id`, updates `collected_fields` / `completion_status`, POSTs the webhook) +- `services/chatbot/agents/onboarding/{graph,prompts,types}.py` — `run_chatbot_turn` (public entry point), `build_onboarding_graph`, the extraction/reply nodes, `FieldExtraction` +- `services/chatbot/agents/rag/graph.py` — `run_rag_turn` (public entry point), `build_rag_graph`, `build_rag_system_prompt`, `MAX_TOOL_ROUNDS` +- `services/chatbot/agents/rag/retriever.py` — `get_knowledge_base_client` (project OpenAI credentials → client; missing → `ChatbotInputError`), `search_knowledge_bases`, `format_passages`, `build_search_tool` +- `services/chatbot/agents/rag/prompts.py` — `RAG_INSTRUCTION`, tool name/description, passage templates +- `services/chatbot/agents/rag/types.py` — `RagState`, `KnowledgeBasePassage`, `SearchKnowledgeBaseArgs` +- `services/chatbot/agents/registry.py` — `run_agent_turn` (dispatches on `ChatbotFlowNodeTypeEnum`, i.e. the active flow node's type; rag needs the OpenAI client) +- `services/chatbot/provider_errors.py` — `to_provider_error`, `CHAT_MODEL_ERRORS` (Anthropic error mapping shared by both agents) +- `services/chatbot/history.py` — agent-agnostic history plumbing: `summarize_if_needed`, `to_langchain_messages`, `extract_final_reply` (empty → `ChatbotProviderError`), `validate_turn_input`, kickoff message, `has_user_turn`, `insert_after_leading_system` +- `services/chatbot/types.py` — shared `ChatbotTurnResult`, `ChatbotSource`, `ChatbotMessageDict` +- `services/chatbot/config.py` — `get_chatbot_config_version` (requested or latest version), `build_chatbot_model_settings` (version row → `ChatbotModelSettings`, incl. `instructions`, `knowledge_base_ids`, `max_num_results`; 422-class `ChatbotInputError` for non-Anthropic/non-text/no-model configs or a bad thinking budget), `get_anthropic_api_key` (project's own credentials; no platform fallback), `resolve_chatbot_model_settings` (all three, used by the worker), `prepare_rag_search_client` (non-empty `knowledge_base_ids`, each a non-deleted OpenAI collection's `knowledge_base_id` via `CollectionCrud.read_all`, plus OpenAI credentials → client) +- `services/chatbot/utils.py` — `ChatbotModelSettings` (incl. `api_key`, kept out of `repr`/`eq`), defaults (`DEFAULT_MAX_OUTPUT_TOKENS`, `DEFAULT_HISTORY_TOKEN_LIMIT`, `FLOW_KEEP_RECENT_MESSAGES`), `get_chat_model` - `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` ## 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"}}`. +- Webhook success payload: `{"success": true, "data": {"session_id", "chatbot_id", "node_id", "reply", "status", "completion_status", "collected_fields", "sources": [{"file_id", "filename"}]}}`; 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) — onboarding: two calls per turn once the user has spoken (extraction + reply); rag: one call per tool round (≤ 4); plus one summary call when history exceeds the token budget. +- OpenAI vector store search (`vector_stores.search`, rag only) with the project's OpenAI credentials, one call per knowledge base per tool call; `OPENAI_TIMEOUT_SECONDS`, one SDK retry. - Caller webhook (`send_callback`, HMAC-signed with the project webhook secret). - kaapi-frontend React Flow editor reads/writes the flow JSON as-is.