feat: upload flow

Author: Tincho44

app/agents/faq.py

@@ -17,6 +17,7 @@
 from ..db.config.database import get_async_session
 from ..graph.state import ConversationState
 from ..services.embedding_service import embedding_service
+from ..services import conversation_service
 
 logger = logging.getLogger(__name__)
 
@@ -106,13 +107,25 @@ async def __call__(self, state: ConversationState) -> ConversationState:
                     "query_processed": True
                 }
 
+                # --- Persistencia en DB ---
+                conv_id = state.get("conversation_id")
+                try:
+                    conv_id = await conversation_service.get_or_create_conversation(
+                        session, conv_id
+                    )
+                    await conversation_service.save_message(session, conv_id, "user", user_message)
+                    await conversation_service.save_message(session, conv_id, "assistant", response)
+                except Exception as db_err:
+                    logger.error(f"[FAQ] Error al persistir mensajes en DB: {db_err}", exc_info=True)
+
                 # Devolver delta de messages para que add_messages lo agregue
                 return {
                     "agent_context": state["agent_context"],
                     "faq_results": state.get("faq_results", []),
                     "next_action": state["next_action"],
                     "should_continue": state["should_continue"],
-                    "messages": [AIMessage(content=response)]
+                    "messages": [AIMessage(content=response)],
+                    "conversation_id": conv_id,
                 }
 
         except Exception as e:

app/agents/wizard_node.py

@@ -15,6 +15,8 @@
 from .base import AgentNode
 from ..graph.state import ConversationState
 from .wizard_workflow.wizard_graph import wizard_graph
+from ..db.config.database import get_async_session
+from ..services import conversation_service
 
 logger = logging.getLogger(__name__)
 
@@ -77,13 +79,51 @@ async def __call__(self, state: ConversationState) -> ConversationState:
         logger.debug(f"[WIZARD_NODE]   wizard_status={result.get('wizard_status')}")
         logger.debug(f"[WIZARD_NODE]   response (first 200 chars): {response_content[:200]!r}")
 
+        # --- Persistencia en DB ---
+        conv_id = state.get("conversation_id")
+        wizard_responses = result.get("wizard_responses", {})
+        email = wizard_responses.get("email")
+
+        try:
+            async for db_session in get_async_session():
+                conv_id = await conversation_service.get_or_create_conversation(
+                    db_session, conv_id, email=email
+                )
+
+                # Guardar mensaje del usuario
+                user_msgs = [m for m in state.get("messages", []) if m.type == "human"]
+                if user_msgs:
+                    await conversation_service.save_message(
+                        db_session, conv_id, "user", user_msgs[-1].content
+                    )
+
+                # Guardar respuesta del asistente
+                if response_content:
+                    await conversation_service.save_message(
+                        db_session, conv_id, "assistant", response_content
+                    )
+
+                # Actualizar WizardSession con los datos del paso actual
+                ws = await conversation_service.get_or_create_wizard_session(db_session, conv_id)
+                status = "COMPLETED" if result.get("completed") else "ACTIVE"
+                await conversation_service.update_wizard_session(
+                    db_session,
+                    ws,
+                    result.get("current_question", 1),
+                    wizard_responses,
+                    status,
+                )
+        except Exception as e:
+            logger.error(f"[WIZARD_NODE] Error al persistir en DB: {e}", exc_info=True)
+
         return {
             **state,
             "wizard_state": result,
             "messages": result.get("messages", []),
             "agent_context": {
                 "response": response_content
             },
+            "conversation_id": conv_id,
         }
 
 

app/agents/wizard_workflow/nodes.py

@@ -24,11 +24,18 @@ def ask_question_node(state: WizardState):
 def store_answer_node(state: WizardState):
     user_message = [m.content for m in state["messages"] if m.type == "human"][-1]
 
-    new_index = state["current_question"] + 1
+    current_q = state["current_question"]
+    new_index = current_q + 1
     is_completed = new_index > max(WIZARD_QUESTIONS.keys())
 
+    q_config = WIZARD_QUESTIONS.get(current_q, {})
+    field_name = q_config.get("field_name", str(current_q))
+
+    wizard_responses = dict(state.get("wizard_responses", {}))
+    wizard_responses[field_name] = user_message
+
     logger.debug("-" * 60)
-    logger.debug(f"[WIZARD/store_answer] Storing answer for question #{state['current_question']}")
+    logger.debug(f"[WIZARD/store_answer] Storing answer for question #{current_q} (field={field_name!r})")
     logger.debug(f"[WIZARD/store_answer] User answer: {user_message[:200]!r}")
     logger.debug(f"[WIZARD/store_answer] Next question index: {new_index}")
     logger.debug(f"[WIZARD/store_answer] Is completed: {is_completed}")
@@ -37,6 +44,7 @@ def store_answer_node(state: WizardState):
     return {
         **state,
         "answers": state.get("answers", []) + [user_message],
+        "wizard_responses": wizard_responses,
         "current_question": new_index,
         "completed": is_completed,
         "awaiting_answer": False,

app/services/conversation_service.py

@@ -0,0 +1,98 @@
+"""
+Servicio para persistir datos de conversaciones en la base de datos.
+
+Centraliza las operaciones de DB relacionadas a conversaciones, mensajes
+y sesiones del wizard. Todos los métodos reciben una sesión activa de
+SQLAlchemy para poder participar de la misma transacción.
+"""
+
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..db.models import Conversation, Message, WizardSession
+
+logger = logging.getLogger(__name__)
+
+
+async def get_or_create_conversation(
+    session: AsyncSession,
+    conv_id: int | None,
+    email: str | None = None,
+) -> int:
+    """Devuelve el conv_id existente o crea una nueva Conversation.
+
+    Si se pasa email y la conversación no lo tiene guardado, lo actualiza.
+    """
+    if conv_id is not None:
+        if email:
+            conv = await session.get(Conversation, conv_id)
+            if conv and not conv.email:
+                conv.email = email
+                await session.commit()
+        return conv_id
+
+    conv = Conversation(email=email)
+    session.add(conv)
+    await session.flush()
+    await session.commit()
+    logger.debug(f"[CONV_SERVICE] Nueva conversación creada id={conv.id}")
+    return conv.id
+
+
+async def save_message(
+    session: AsyncSession,
+    conv_id: int,
+    role: str,
+    content: str,
+) -> None:
+    """Guarda un mensaje en la tabla messages."""
+    msg = Message(conv_id=conv_id, role=role, content=content)
+    session.add(msg)
+    await session.commit()
+    logger.debug(f"[CONV_SERVICE] Mensaje guardado role={role!r} conv_id={conv_id}")
+
+
+async def get_or_create_wizard_session(
+    session: AsyncSession,
+    conv_id: int,
+) -> WizardSession:
+    """Obtiene la WizardSession activa para la conversación o crea una nueva."""
+    result = await session.execute(
+        select(WizardSession)
+        .where(WizardSession.conv_id == conv_id)
+        .where(WizardSession.state == "ACTIVE")
+        .order_by(WizardSession.created_at.desc())
+        .limit(1)
+    )
+    ws = result.scalar_one_or_none()
+    if ws is None:
+        ws = WizardSession(
+            conv_id=conv_id,
+            current_question=1,
+            responses={},
+            state="ACTIVE",
+        )
+        session.add(ws)
+        await session.flush()
+        logger.debug(f"[CONV_SERVICE] Nueva WizardSession creada conv_id={conv_id}")
+    return ws
+
+
+async def update_wizard_session(
+    session: AsyncSession,
+    wizard_session: WizardSession,
+    current_question: int,
+    responses: dict,
+    status: str,
+) -> None:
+    """Actualiza current_question, responses y state de la WizardSession."""
+    wizard_session.current_question = current_question
+    wizard_session.responses = responses
+    wizard_session.state = status
+    await session.commit()
+    logger.debug(
+        f"[CONV_SERVICE] WizardSession actualizada id={wizard_session.id} "
+        f"question={current_question} status={status!r}"
+    )