init

.gitignore

+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual Environment
+.venv/
+venv/
+ENV/
+
+# IDEs
+.vscode/
+.idea/
+
+# Backend specifically
+backend/.env
+backend/.venv/
+backend/__pycache__/
+backend/*.pyc
+backend/.pyscn/
+
+# Preference folder (development/temporary)
+preference/
+
+# Development/Test folders
+backend/hehe/
+backend/test/
+backend/scripts/
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Misc
+.ruff_cache/
+*.log
+!backend/requirements.txt
+run.txt

.gitlab-ci.yml

+stages:
+  - deploy
+
+deploy_to_server:
+  stage: deploy
+  # Chế độ shell chạy trực tiếp trên host
+  script:
+    - echo "🚀 Bắt đầu quá trình Deploy Sạch..."
+    - export DOCKER_BUILDKIT=1
+    - "if [ -f /home/anhvh/canifa_soure/chatbot-canifa/backend/.env ]; then cp /home/anhvh/canifa_soure/chatbot-canifa/backend/.env backend/.env; fi"
+    - cd backend
+    - docker stop canifa_backend || true
+    - docker rm -f canifa_backend || true
+    - docker compose -p canifa-chatbot down --remove-orphans || true
+    - docker compose -p canifa-chatbot up --build -d
+    - docker system prune -f
+    - echo "✅ Web ĐÃ CẬP NHẬT THÀNH CÔNG!"
+  only:
+    - master

.pre-commit-config.yaml

+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.8.4
+    hooks:
+      # Run the linter.
+      - id: ruff
+        args: [ --fix ]
+      # Run the formatter.
+      - id: ruff-format

backend/.dockerignore

+__pycache__
+*.pyc
+.env
+.venv
+venv
+.git
+.gitignore
+.dockerignore
+logs
+data

backend/.editorconfig

+# EditorConfig - đảm bảo indentation nhất quán
+# https://editorconfig.org
+
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.py]
+indent_style = space
+indent_size = 4
+max_line_length = 120
+
+[*.{json,yml,yaml}]
+indent_style = space
+indent_size = 2
+
+[*.md]
+trim_trailing_whitespace = false
+

backend/.gitignore

+
+# Ignore embedded repo
+preference/
+

backend/5.10.0

+Requirement already satisfied: python-socketio in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (5.13.0)
+Requirement already satisfied: bidict>=0.21.0 in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (from python-socketio) (0.23.1)
+Requirement already satisfied: python-engineio>=4.11.0 in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (from python-socketio) (4.12.2)
+Requirement already satisfied: simple-websocket>=0.10.0 in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (from python-engineio>=4.11.0->python-socketio) (1.1.0)
+Requirement already satisfied: wsproto in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (from simple-websocket>=0.10.0->python-engineio>=4.11.0->python-socketio) (1.2.0)
+Requirement already satisfied: h11<1,>=0.9.0 in c:\users\fptshop\miniconda3\envs\robot\lib\site-packages (from wsproto->simple-websocket>=0.10.0->python-engineio>=4.11.0->python-socketio) (0.16.0)

backend/Dockerfile.dev

+# ============================================================
+# DOCKERFILE.DEV - Local Development (Hot Reload + Cache)
+# ============================================================
+# Sử dụng Python 3.11 Slim để tối ưu dung lượng
+FROM python:3.11-slim
+
+# Thiết lập thư mục làm việc
+WORKDIR /app
+
+# Thiết lập biến môi trường
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV ENV=development
+
+# Copy requirements.txt trước để tận dụng Docker cache
+COPY requirements.txt .
+
+# Cài đặt thư viện Python (Docker layer cache)
+RUN pip install -r requirements.txt && pip install watchdog[watchmedo]
+
+# Copy toàn bộ source code vào image
+COPY . .
+
+# Expose port 5000
+EXPOSE 5000
+
+# Health check (optional)
+HEALTHCHECK --interval=10s --timeout=5s --start-period=5s --retries=2 \
+    CMD python -c "import requests; requests.get('http://localhost:5000/docs')" || exit 1
+
+CMD ["gunicorn", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:5000", "--timeout", "120", "--reload", "server:app"]

backend/Dockerfile.prod

+# Lấy Python 3.11 slim (ít file, nhẹ)
+FROM python:3.11-slim
+
+# Thư mục làm việc
+WORKDIR /app
+
+ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
+
+# Copy requirements rồi cài package
+COPY requirements.txt .
+RUN pip install -r requirements.txt
+
+# Copy code
+COPY . .
+
+# Copy entrypoint script (nếu có)
+COPY entrypoint.sh /app/entrypoint.sh
+RUN chmod +x /app/entrypoint.sh
+
+# Mở port 5000
+EXPOSE 5000
+
+# Chạy server
+CMD ["/app/entrypoint.sh"]
\ No newline at end of file

backend/Dockerfile.stage

+# ============================================================
+# DOCKERFILE.STAGE - Development/Staging (Hot Reload)
+# ============================================================
+# Sử dụng Python 3.11 Slim để tối ưu dung lượng
+FROM python:3.11-slim
+
+# Thiết lập thư mục làm việc
+WORKDIR /app
+
+# Thiết lập biến môi trường để log in ra ngay lập tức
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV ENV=development
+
+# Copy requirements.txt trước để tận dụng Docker cache
+COPY requirements.txt .
+
+# Cài đặt thư viện Python (với cache mount)
+RUN pip install -r requirements.txt
+
+# Copy toàn bộ source code vào image
+COPY . .
+
+# Expose port 5000 (Port chạy server)
+EXPOSE 5000
+
+# Lệnh chạy server dùng uvicorn với hot reload
+# ⚡ Hot reload - tự động restart khi code thay đổi (dùng cho dev/stage)
+CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "5000", "--reload"]

backend/Makefile

+# Makefile cho CANIFA Chatbot
+
+.PHONY: up down restart logs build ps clean setup-nginx monitor-up monitor-down build-dev run-dev
+
+up:
+	sudo docker compose up -d --build
+
+down:
+	docker-compose down
+
+restart:
+	docker-compose restart backend
+
+logs:
+	sudo 
+ps:
+	docker-compose ps
+
+build:
+	docker-compose build
+
+build-dev:
+	docker build -f Dockerfile.dev -t canifa-backend:dev .
+
+run-dev:
+	docker run -it --rm -v $(PWD):/app -p 5000:5000 canifa-backend:dev
+
+clean:
+	docker-compose down -v --rmi all --remove-orphans
+
+setup-nginx:
+	@echo "🚀 Đang cấu hình Nginx..."
+	sudo cp nginx.conf /etc/nginx/sites-available/chatbot
+	sudo ln -sf /etc/nginx/sites-available/chatbot /etc/nginx/sites-enabled/
+	sudo nginx -t && sudo systemctl restart nginx
+	@echo "✅ Nginx đã được cấu hình và restart!"
+

backend/__init__.py

+"""
+Backend Package
+"""

backend/agent/__init__.py

+"""
+Fashion Q&A Agent Package
+"""
+
+from .graph import build_graph
+from .models import AgentConfig, AgentState, get_config
+
+__all__ = [
+    "AgentConfig",
+    "AgentState",
+    "build_graph",
+    "get_config",
+]

backend/agent/controller.py

+"""
+Fashion Q&A Agent Controller
+Langfuse will auto-trace via LangChain integration (no code changes needed).
+"""
+
+import logging
+import time
+import uuid
+
+from fastapi import BackgroundTasks
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.runnables import RunnableConfig
+
+from common.cache import redis_cache
+from common.conversation_manager import get_conversation_manager
+from common.langfuse_client import get_callback_handler
+from common.llm_factory import create_llm
+from config import DEFAULT_MODEL, REDIS_CACHE_TURN_ON
+from langfuse import propagate_attributes
+
+from .graph import build_graph
+from .helper import extract_product_ids, handle_post_chat_async, parse_ai_response
+from .models import AgentState, get_config
+from .tools.get_tools import get_all_tools
+
+logger = logging.getLogger(__name__)
+
+
+async def chat_controller(
+    query: str,
+    user_id: str,
+    background_tasks: BackgroundTasks,
+    model_name: str = DEFAULT_MODEL,
+    images: list[str] | None = None,
+    identity_key: str | None = None,
+) -> dict:
+    """
+    Controller main logic for non-streaming chat requests.
+
+    Flow:
+    1. Check cache (if enabled) → HIT: return cached response
+    2. MISS: Call LLM → Save to cache → Return response
+    
+    Args:
+        identity_key: Key for saving/loading history (identity.history_key)
+                      Guest: device_id, User: user_id
+    """
+    effective_identity_key = identity_key or user_id
+    
+    logger.info(
+        "chat_controller start: model=%s, user_id=%s, identity_key=%s", 
+        model_name, user_id, effective_identity_key
+    )
+
+    # ====================== CACHE LAYER ======================
+    if REDIS_CACHE_TURN_ON:
+        cached_response = await redis_cache.get_response(
+            user_id=effective_identity_key, query=query
+        )
+        if cached_response:
+            logger.info(f"⚡ CACHE HIT for identity_key={effective_identity_key}")
+            memory = await get_conversation_manager()
+            background_tasks.add_task(
+                handle_post_chat_async,
+                memory=memory,
+                identity_key=effective_identity_key,
+                human_query=query,
+                ai_response=cached_response,
+            )
+            return {**cached_response, "cached": True}
+
+    # ====================== NORMAL LLM FLOW ======================
+    logger.info("chat_controller: proceed with live LLM call")
+
+    config = get_config()
+    config.model_name = model_name
+
+    llm = create_llm(model_name=model_name, streaming=False, json_mode=True)
+    tools = get_all_tools()
+    graph = build_graph(config, llm=llm, tools=tools)
+
+    # Init ConversationManager (Singleton)
+    memory = await get_conversation_manager()
+
+    # Load History
+    history_dicts = await memory.get_chat_history(effective_identity_key, limit=20)
+    messages = [
+        HumanMessage(content=m["message"]) if m["is_human"] else AIMessage(content=m["message"])
+        for m in history_dicts
+    ]
+
+    # Prepare State
+    initial_state: AgentState = {
+        "user_query": HumanMessage(content=query),
+        "messages": messages + [HumanMessage(content=query)],
+        "history": messages,
+        "user_id": user_id,
+        "images_embedding": [],
+        "ai_response": None,
+    }
+    
+    run_id = str(uuid.uuid4())
+    langfuse_handler = get_callback_handler()
+    
+    exec_config = RunnableConfig(
+        configurable={
+            "user_id": user_id,
+            "transient_images": images or [],
+            "run_id": run_id,
+        },
+        run_id=run_id,
+        metadata={"run_id": run_id, "tags": "chatbot,production"},
+        callbacks=[langfuse_handler] if langfuse_handler else [],
+    )
+
+    # Execute Graph
+    start_time = time.time()
+    session_id = f"{user_id}-{run_id[:8]}"
+    
+    with propagate_attributes(user_id=user_id, session_id=session_id):
+        result = await graph.ainvoke(initial_state, config=exec_config)
+    
+    duration = time.time() - start_time
+
+    # Parse Response
+    all_product_ids = extract_product_ids(result.get("messages", []))
+    ai_raw_content = result.get("ai_response").content if result.get("ai_response") else ""
+    ai_text_response, final_product_ids = parse_ai_response(ai_raw_content, all_product_ids)
+
+    response_payload = {
+        "ai_response": ai_text_response,
+        "product_ids": final_product_ids,
+    }
+
+    # ====================== SAVE TO CACHE ======================
+    if REDIS_CACHE_TURN_ON:
+        await redis_cache.set_response(
+            user_id=effective_identity_key, 
+            query=query, 
+            response_data=response_payload, 
+            ttl=300
+        )
+        logger.debug(f"💾 Cached response for identity_key={effective_identity_key}")
+
+    # Save to History (Background)
+    background_tasks.add_task(
+        handle_post_chat_async,
+        memory=memory,
+        identity_key=effective_identity_key,
+        human_query=query,
+        ai_response=response_payload,
+    )
+
+    logger.info("chat_controller finished in %.2fs", duration)
+    return {**response_payload, "cached": False}
+

backend/agent/graph.py

+"""
+Fashion Q&A Agent Graph
+LangGraph workflow với clean architecture.
+Tất cả resources (LLM, Tools) khởi tạo trong __init__.
+Sử dụng ConversationManager (Postgres) để lưu history thay vì checkpoint.
+"""
+
+import logging
+from typing import Any
+
+from langchain_core.language_models import BaseChatModel
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.runnables import RunnableConfig
+from langgraph.cache.memory import InMemoryCache
+from langgraph.graph import END, StateGraph
+from langgraph.prebuilt import ToolNode
+from langgraph.types import CachePolicy
+
+from common.llm_factory import create_llm
+
+from .models import AgentConfig, AgentState, get_config
+from .prompt import get_system_prompt
+from .tools.get_tools import get_all_tools, get_collection_tools
+
+logger = logging.getLogger(__name__)
+
+
+class CANIFAGraph:
+    """
+    Fashion Q&A Agent Graph Manager.
+    """
+
+    def __init__(
+        self,
+        config: AgentConfig | None = None,
+        llm: BaseChatModel | None = None,
+        tools: list | None = None,
+    ):
+        self.config = config or get_config()
+        self._compiled_graph: Any | None = None
+        self.llm: BaseChatModel = llm or create_llm(
+            model_name=self.config.model_name, api_key=self.config.openai_api_key, streaming=True
+        )
+        self.all_tools = tools or get_all_tools()
+        self.collection_tools = get_collection_tools()  # Vẫn lấy list name để routing
+        self.retrieval_tools = self.all_tools
+
+        self.llm_with_tools = self.llm.bind_tools(self.all_tools, strict=True)
+        self.system_prompt = get_system_prompt()
+        self.prompt_template = ChatPromptTemplate.from_messages(
+            [
+                ("system", self.system_prompt),
+                MessagesPlaceholder(variable_name="history"),
+                MessagesPlaceholder(variable_name="user_query"),
+                MessagesPlaceholder(variable_name="messages"),
+            ]
+        )
+        self.chain = self.prompt_template | self.llm_with_tools
+        self.cache = InMemoryCache()
+
+    async def _agent_node(self, state: AgentState, config: RunnableConfig) -> dict:
+        """Agent node - Chỉ việc đổ dữ liệu riêng vào khuôn đã có sẵn."""
+        messages = state.get("messages", [])
+        history = state.get("history", [])
+        user_query = state.get("user_query")
+
+        transient_images = config.get("configurable", {}).get("transient_images", [])
+        if transient_images and messages:
+            pass
+        # Invoke chain with user_query, history, and messages
+        response = await self.chain.ainvoke({
+            "user_query": [user_query] if user_query else [],
+            "history": history,
+            "messages": messages
+        })
+        return {"messages": [response], "ai_response": response}
+
+
+    def _should_continue(self, state: AgentState) -> str:
+        """Routing: tool nodes hoặc end."""
+        last_message = state["messages"][-1]
+
+        if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
+            logger.info("🏁 Agent finished")
+            return "end"
+
+        tool_names = [tc["name"] for tc in last_message.tool_calls]
+        collection_names = [t.name for t in self.collection_tools]
+
+        if any(name in collection_names for name in tool_names):
+            logger.info(f"🔄 → collect_tools: {tool_names}")
+            return "collect_tools"
+
+        logger.info(f"🔄 → retrieve_tools: {tool_names}")
+        return "retrieve_tools"
+
+    def build(self) -> Any:
+        """Build và compile LangGraph workflow."""
+        if self._compiled_graph is not None:
+            return self._compiled_graph
+        workflow = StateGraph(AgentState)
+
+        # Nodes
+        workflow.add_node("agent", self._agent_node)
+        workflow.add_node("retrieve_tools", ToolNode(self.retrieval_tools), cache_policy=CachePolicy(ttl=3600))
+        workflow.add_node("collect_tools", ToolNode(self.collection_tools))
+
+        # Edges
+        workflow.set_entry_point("agent")
+        workflow.add_conditional_edges(
+            "agent",
+            self._should_continue,
+            {"retrieve_tools": "retrieve_tools", "collect_tools": "collect_tools", "end": END},
+        )
+        workflow.add_edge("retrieve_tools", "agent")
+        workflow.add_edge("collect_tools", "agent")
+
+        self._compiled_graph = workflow.compile(cache=self.cache)  # No Checkpointer
+        logger.info("✅ Graph compiled (Langfuse callback will be per-run)")
+
+        return self._compiled_graph
+
+    @property
+    def graph(self) -> Any:
+        return self.build()
+
+
+# --- Singleton & Public API ---
+_instance: list[CANIFAGraph | None] = [None]
+
+def build_graph(config: AgentConfig | None = None, llm: BaseChatModel | None = None, tools: list | None = None) -> Any:
+    """Get compiled graph (singleton)."""
+    if _instance[0] is None:
+        _instance[0] = CANIFAGraph(config, llm, tools)
+    return _instance[0].build()
+
+
+def get_graph_manager(
+    config: AgentConfig | None = None, llm: BaseChatModel | None = None, tools: list | None = None
+) -> CANIFAGraph:
+    """Get CANIFAGraph instance."""
+    if _instance[0] is None:
+        _instance[0] = CANIFAGraph(config, llm, tools)
+    return _instance[0]
+
+
+def reset_graph() -> None:
+    """Reset singleton for testing."""
+    _instance[0] = None

backend/agent/helper.py

+"""
+Agent Helper Functions
+Các hàm tiện ích cho chat controller.
+"""
+
+import json
+import logging
+import uuid
+
+from langchain_core.messages import HumanMessage, ToolMessage
+from langchain_core.runnables import RunnableConfig
+
+from common.conversation_manager import ConversationManager
+from common.langfuse_client import get_callback_handler
+from .models import AgentState
+
+logger = logging.getLogger(__name__)
+
+
+def extract_product_ids(messages: list) -> list[dict]:
+    """
+    Extract full product info from tool messages (data_retrieval_tool results).
+    Returns list of product objects with: sku, name, price, sale_price, url, thumbnail_image_url.
+    """
+    products = []
+    seen_skus = set()
+
+    for msg in messages:
+        if isinstance(msg, ToolMessage):
+            try:
+                # Tool result is JSON string
+                tool_result = json.loads(msg.content)
+
+                # Check if tool returned products
+                if tool_result.get("status") == "success" and "products" in tool_result:
+                    for product in tool_result["products"]:
+                        sku = product.get("internal_ref_code")
+                        if sku and sku not in seen_skus:
+                            seen_skus.add(sku)
+
+                            # Extract full product info
+                            product_obj = {
+                                "sku": sku,
+                                "name": product.get("magento_product_name", ""),
+                                "price": product.get("price_vnd", 0),
+                                "sale_price": product.get("sale_price_vnd"),  # null nếu không sale
+                                "url": product.get("magento_url_key", ""),
+                                "thumbnail_image_url": product.get("thumbnail_image_url", ""),
+                            }
+                            products.append(product_obj)
+            except (json.JSONDecodeError, KeyError, TypeError) as e:
+                logger.debug(f"Could not parse tool message for products: {e}")
+                continue
+
+    return products
+
+
+def parse_ai_response(ai_raw_content: str, all_product_ids: list) -> tuple[str, list]:
+    """
+    Parse AI response từ LLM output.
+    
+    Args:
+        ai_raw_content: Raw content từ AI response
+        all_product_ids: Product IDs extracted từ tool messages
+        
+    Returns:
+        tuple: (ai_text_response, final_product_ids)
+    """
+    ai_text_response = ai_raw_content
+    final_product_ids = all_product_ids
+
+    try:
+        # Try to parse if it's a JSON string from LLM
+        ai_json = json.loads(ai_raw_content)
+        ai_text_response = ai_json.get("ai_response", ai_raw_content)
+        explicit_ids = ai_json.get("product_ids", [])
+        if explicit_ids and isinstance(explicit_ids, list):
+            # Replace with explicit IDs from LLM
+            final_product_ids = explicit_ids
+    except (json.JSONDecodeError, TypeError):
+        pass
+
+    return ai_text_response, final_product_ids
+
+
+def prepare_execution_context(query: str, user_id: str, history: list, images: list | None):
+    """
+    Prepare initial state and execution config for the graph run.
+    
+    Returns:
+        tuple: (initial_state, exec_config)
+    """
+    initial_state: AgentState = {
+        "user_query": HumanMessage(content=query),
+        "messages": [HumanMessage(content=query)],
+        "history": history,
+        "user_id": user_id,
+        "images_embedding": [],
+        "ai_response": None,
+    }
+    run_id = str(uuid.uuid4())
+
+    # Metadata for LangChain (tags for logging/filtering)
+    metadata = {
+        "run_id": run_id,
+        "tags": "chatbot,production",
+    }
+
+    langfuse_handler = get_callback_handler()
+
+    exec_config = RunnableConfig(
+        configurable={
+            "user_id": user_id,
+            "transient_images": images or [],
+            "run_id": run_id,
+        },
+        run_id=run_id,
+        metadata=metadata,
+        callbacks=[langfuse_handler] if langfuse_handler else [],
+    )
+    return initial_state, exec_config
+
+
+async def handle_post_chat_async(
+    memory: ConversationManager, 
+    identity_key: str, 
+    human_query: str, 
+    ai_response: dict | None
+):
+    """
+    Save chat history in background task after response is sent.
+    Lưu AI response dưới dạng JSON string.
+    """
+    if ai_response:
+        try:
+            # Convert dict thành JSON string để lưu vào TEXT field
+            ai_response_json = json.dumps(ai_response, ensure_ascii=False)
+            await memory.save_conversation_turn(identity_key, human_query, ai_response_json)
+            logger.debug(f"Saved conversation for identity_key {identity_key}")
+        except Exception as e:
+            logger.error(f"Failed to save conversation for identity_key {identity_key}: {e}", exc_info=True)

backend/agent/models.py

+from typing import Annotated, Any, TypedDict
+
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+from pydantic import BaseModel
+
+import config as global_config
+
+
+class QueryRequest(BaseModel):
+    """API Request model cho Fashion Q&A Chat"""
+
+    user_id: str | None = None
+    user_query: str
+    images: list[str] | None = None
+    image_analysis: dict[str, Any] | None = None
+
+
+class AgentState(TypedDict):
+    """Trạng thái của Agent trong LangGraph."""
+
+    user_query: BaseMessage
+    history: list[BaseMessage]
+    user_id: str | None
+    ai_response: BaseMessage | None
+    images_embedding: list[str] | None
+    messages: Annotated[list[BaseMessage], add_messages]
+
+
+class AgentConfig:
+    """Class chứa cấu hình runtime cho Agent."""
+
+    def __init__(self, **kwargs):
+        self.model_name = kwargs.get("model_name") or global_config.DEFAULT_MODEL
+        self.openai_api_key = kwargs.get("openai_api_key")
+        self.google_api_key = kwargs.get("google_api_key")
+        self.groq_api_key = kwargs.get("groq_api_key")
+        self.supabase_url = kwargs.get("supabase_url")
+        self.supabase_key = kwargs.get("supabase_key")
+        self.langfuse_public_key = kwargs.get("langfuse_public_key")
+        self.langfuse_secret_key = kwargs.get("langfuse_secret_key")
+        self.langfuse_base_url = kwargs.get("langfuse_base_url")
+
+
+def get_config() -> AgentConfig:
+    """Khởi tạo cấu hình Agent từ các biến môi trường."""
+    return AgentConfig(
+        model_name=global_config.DEFAULT_MODEL,
+        openai_api_key=global_config.OPENAI_API_KEY,
+        google_api_key=global_config.GOOGLE_API_KEY,
+        groq_api_key=global_config.GROQ_API_KEY,
+        supabase_url=global_config.AI_SUPABASE_URL,
+        supabase_key=global_config.AI_SUPABASE_KEY,
+        langfuse_public_key=global_config.LANGFUSE_PUBLIC_KEY,
+        langfuse_secret_key=global_config.LANGFUSE_SECRET_KEY,
+        langfuse_base_url=global_config.LANGFUSE_BASE_URL,
+    )

backend/agent/nodes/__init__.py

+"""
+Agent Nodes Package
+"""
+
+from .agent import agent_node
+
+__all__ = ["agent_node"]

backend/agent/prompt.py

+"""
+CiCi Fashion Consultant - System Prompt
+Tư vấn thời trang CANIFA chuyên nghiệp
+Version 3.0 - Dynamic from File
+"""
+
+import os
+from datetime import datetime
+
+PROMPT_FILE_PATH = os.path.join(os.path.dirname(__file__), "system_prompt.txt")
+
+def get_system_prompt() -> str:
+    """
+    System prompt cho CiCi Fashion Agent
+    Đọc từ file system_prompt.txt để có thể update dynamic.
+    
+    Returns:
+        str: System prompt với ngày hiện tại
+    """
+    now = datetime.now()
+    date_str = now.strftime("%d/%m/%Y")
+
+    try:
+        if os.path.exists(PROMPT_FILE_PATH):
+            with open(PROMPT_FILE_PATH, "r", encoding="utf-8") as f:
+                prompt_template = f.read()
+                return prompt_template.replace("{date_str}", date_str)
+    except Exception as e:
+        print(f"Error reading system prompt file: {e}")
+
+    # Fallback default prompt if file error
+    return f"""# VAI TRÒ
+Bạn là CiCi - Chuyên viên tư vấn thời trang CANIFA.
+Hôm nay: {date_str}
+
+KHÔNG BAO GIỜ BỊA ĐẶT. TRẢ LỜI NGẮN GỌN.
+"""
\ No newline at end of file

backend/agent/system_prompt.txt

+# VAI TRÒ
+
+Bạn là CiCi - Chuyên viên tư vấn thời trang CANIFA.
+- Nhiệt tình, thân thiện, chuyên nghiệp
+- CANIFA BÁN QUẦN ÁO: áo, quần, váy, đầm, phụ kiện thời trang
+- Hôm nay: {date_str}
+
+---
+
+# QUY TẮC TRUNG THỰC - BẮT BUỘC
+
+KHÔNG BAO GIỜ BỊA ĐẶT - CHỈ NÓI THEO DỮ LIỆU
+
+**ĐÚNG:**
+- Tool trả về áo thun → Giới thiệu áo thun
+- Tool trả về 0 sản phẩm → Nói "Shop chưa có sản phẩm này"
+- Tool trả về quần nỉ mà khách hỏi bikini → Nói "Shop chưa có bikini"
+
+**CẤM:**
+- Tool trả về quần nỉ → Gọi là "đồ bơi"
+- Tool trả về 0 kết quả → Nói "shop có sản phẩm X"
+- Tự bịa mã sản phẩm, giá tiền, chính sách
+
+Không có trong data = Không nói = Không tư vấn láo
+
+---
+
+# NGÔN NGỮ & XƯNG HÔ
+
+- Mặc định: Xưng "mình" - gọi "bạn"
+- Khi khách xưng anh/chị: Xưng "em" - gọi "anh/chị"
+- Khách nói tiếng Việt → Trả lời tiếng Việt
+- Khách nói tiếng Anh → Trả lời tiếng Anh
+- Ngắn gọn, đi thẳng vào vấn đề
+
+---
+
+# KHI NÀO GỌI TOOL
+
+**Gọi data_retrieval_tool khi:**
+- Khách tìm sản phẩm: "Tìm áo...", "Có màu gì..."
+- Khách hỏi sản phẩm cụ thể: "Mã 8TS24W001 có không?"
+- Tư vấn phong cách: "Mặc gì đi cưới?", "Đồ công sở?"
+
+**⚠️ QUY TẮC SINH QUERY (BẮT BUỘC):**
+- **Query chỉ chứa MÔ TẢ SẢN PHẨM** (tên, chất liệu, màu, phong cách).
+- **TUYỆT ĐỐI KHÔNG đưa giá tiền vào chuỗi `query`**.
+- Giá tiền phải đưa vào tham số riêng: `price_min`, `price_max`.
+
+Ví dụ ĐÚNG:
+- Query: "Áo thun nam cotton thoáng mát basic"
+- Price_max: 300000
+
+Ví dụ SAI (Cấm):
+- Query: "Áo thun nam giá dưới 300k" (SAI vì có giá trong query)
+
+**Gọi canifa_knowledge_search khi:**
+- Hỏi chính sách: freeship, đổi trả, bảo hành
+- Hỏi thương hiệu: Canifa là gì, lịch sử
+- Tìm cửa hàng: địa chỉ, giờ mở cửa
+
+**Không gọi tool khi:**
+- Chào hỏi đơn giản: "Hi", "Hello"
+- Hỏi lại về sản phẩm vừa show
+
+---
+
+# XỬ LÝ KẾT QUẢ TỪ TOOL
+
+## Sau khi gọi tool, kiểm tra kết quả:
+
+**Trường hợp 1: CÓ sản phẩm phù hợp (đúng loại, đúng yêu cầu)**
+- DỪNG LẠI, giới thiệu sản phẩm
+- KHÔNG GỌI TOOL LẦN 2
+
+**Trường hợp 2: CÓ kết quả NHƯNG SAI LOẠI**
+
+Ví dụ: Khách hỏi bikini, tool trả về quần nỉ
+
+→ Trả lời thẳng:
+"Dạ shop chưa có bikini ạ. Shop chuyên về quần áo thời trang (áo, quần, váy). Bạn có muốn tìm sản phẩm nào khác không?"
+
+CẤM TUYỆT ĐỐI:
+- Giới thiệu quần nỉ như thể nó là bikini
+- Nói "shop có đồ bơi này bạn tham khảo" khi thực tế là áo/quần thường
+
+**Trường hợp 3: KHÔNG CÓ kết quả (count = 0)**
+- Thử lại 1 LẦN với filter rộng hơn
+- Nếu vẫn không có:
+
+"Dạ shop chưa có sản phẩm [X] ạ. Bạn có thể tham khảo [loại gần nhất] hoặc ghé shop sau nhé!"
+
+---
+
+# FORMAT ĐẦU RA
+
+Trả về JSON (KHÔNG có markdown backticks):
+
+```json
+{{
+    "ai_response": "Câu trả lời ngắn gọn, mô tả bằng [SKU]",
+    "product_ids": [
+        {{
+            "sku": "8TS24W001",
+            "name": "Áo thun nam basic",
+            "price": 200000,
+            "sale_price": 160000,
+            "url": "https://canifa.com/...",
+            "thumbnail_image_url": "https://..."
+        }}
+    ]
+}}
+```
+
+**Quy tắc ai_response:**
+- Mô tả ngắn gọn, nhắc sản phẩm bằng [SKU]
+- Nói qua giá, chất liệu, điểm nổi bật
+- KHÔNG tạo bảng markdown
+- KHÔNG đưa link, ảnh (frontend tự render)
+
+---
+
+# VÍ DỤ
+
+## Example 1: Chào hỏi
+Input: "Chào shop"
+Output:
+```json
+{{
+    "ai_response": "Chào bạn! Mình là CiCi, tư vấn thời trang CANIFA. Mình có thể giúp gì cho bạn?",
+    "product_ids": []
+}}
+```
+
+## Example 2: Tìm sản phẩm CÓ
+Input: "Tìm áo thun nam dưới 300k"
+Tool trả về: 2 sản phẩm áo thun phù hợp
+Output:
+```json
+{{
+    "ai_response": "Shop có 2 mẫu áo thun nam giá dưới 300k:
+
+- [8TS24W009]: Áo thun cotton basic, giá 250k đang sale 200k
+- [6TN24W012]: Áo thun trơn thoải mái, giá 280k
+
+Bạn kéo xuống xem ảnh nhé!",
+    "product_ids": [
+        {{"sku": "8TS24W009", "name": "Áo thun cotton basic", "price": 250000, "sale_price": 200000, "url": "...", "thumbnail_image_url": "..."}},
+        {{"sku": "6TN24W012", "name": "Áo thun trơn", "price": 280000, "sale_price": null, "url": "...", "thumbnail_image_url": "..."}}
+    ]
+}}
+```
+
+## Example 3: Khách hỏi KHÔNG CÓ trong kho
+Input: "Shop có bikini không?"
+Tool trả về: 0 sản phẩm
+Output:
+```json
+{{
+    "ai_response": "Dạ shop chưa có bikini ạ. CANIFA chuyên về quần áo thời trang như áo, quần, váy, đầm. Bạn có muốn tìm mẫu nào khác không?",
+    "product_ids": []
+}}
+```
+
+## Example 4: Tool trả về SAI LOẠI
+Input: "Cho tôi xem đồ bơi"
+Tool trả về: Quần nỉ, áo nỉ (SAI HOÀN TOÀN so với đồ bơi)
+Output:
+```json
+{{
+    "ai_response": "Dạ shop chưa có đồ bơi ạ. Shop chuyên bán quần áo thời trang (áo, quần, váy, áo khoác). Bạn có muốn tìm loại sản phẩm nào khác không?",
+    "product_ids": []
+}}
+```
+
+TUYỆT ĐỐI KHÔNG giới thiệu sản phẩm sai loại
+
+## Example 5: Khách xưng anh/chị
+Input: "Chào em, anh muốn tìm áo sơ mi"
+Output:
+```json
+{{
+    "ai_response": "Chào anh ạ! Em là CiCi. Anh đang tìm áo sơ mi dài tay hay ngắn tay ạ? Để em tư vấn mẫu phù hợp nhất cho anh nhé!",
+    "product_ids": []
+}}
+```
+
+---
+
+# TÓM TẮT
+
+1. CANIFA bán quần áo (áo, quần, váy, đầm, phụ kiện)
+2. Không có trong data = Không nói
+3. Kiểm tra kỹ tên sản phẩm trước khi giới thiệu
+4. Nếu sai loại → Nói thẳng "shop chưa có X"
+5. Không bịa giá, mã sản phẩm, chính sách
+6. Có kết quả phù hợp = DỪNG, không gọi tool lần 2
+7. Trả lời ngắn gọn, dựa 100% vào dữ liệu tool trả về
+
+---
+
+Luôn thành thật, khéo léo, và chuyên nghiệp.
\ No newline at end of file

backend/agent/tools/__init__.py

+"""
+Tools Package
+Export tool và factory function
+"""
+
+from .data_retrieval_tool import data_retrieval_tool
+from .get_tools import get_all_tools
+
+__all__ = ["data_retrieval_tool", "get_all_tools"]

backend/agent/tools/brand_knowledge_tool.py

+import logging
+
+from langchain_core.tools import tool
+from pydantic import BaseModel, Field
+
+from common.embedding_service import create_embedding_async
+from common.starrocks_connection import get_db_connection
+
+logger = logging.getLogger(__name__)
+
+
+class KnowledgeSearchInput(BaseModel):
+    query: str = Field(
+        description="Câu hỏi hoặc nhu cầu tìm kiếm thông tin phi sản phẩm của khách hàng (ví dụ: tìm cửa hàng, hỏi chính sách, tra bảng size...)"
+    )
+
+
+@tool("canifa_knowledge_search", args_schema=KnowledgeSearchInput)
+async def canifa_knowledge_search(query: str) -> str:
+    """
+    Tra cứu TOÀN BỘ thông tin về thương hiệu và dịch vụ của Canifa.
+    Sử dụng tool này khi khách hàng hỏi về:
+
+    1. THƯƠNG HIỆU & GIỚI THIỆU: Lịch sử hình thành, giá trị cốt lõi, sứ mệnh.
+    2. HỆ THỐNG CỬA HÀNG: Tìm địa chỉ, số điện thoại, giờ mở cửa các cửa hàng tại các tỉnh thành (Hà Nội, HCM, Đà Nẵng, v.v.).
+    3. CHÍNH SÁCH BÁN HÀNG: Quy định đổi trả, bảo hành, chính sách vận chuyển, phí ship.
+    4. KHÁCH HÀNG THÂN THIẾT (KHTT): Điều kiện đăng ký thành viên, các hạng thẻ (Green, Silver, Gold, Diamond), quyền lợi tích điểm, thẻ quà tặng.
+    5. HỖ TRỢ & FAQ: Giải đáp thắc mắc thường gặp, chính sách bảo mật, thông tin liên hệ văn phòng, tuyển dụng.
+    6. TRA CỨU SIZE (BẢNG KÍCH CỠ): Hướng dẫn chọn size chuẩn cho nam, nữ, trẻ em dựa trên chiều cao, cân nặng.
+
+    Ví dụ các câu hỏi phù hợp:
+    - 'Canifa ở Cầu Giấy địa chỉ ở đâu?'
+    - 'Chính sách đổi trả hàng trong bao nhiêu ngày?'
+    - 'Làm sao để lên hạng thẻ Gold?'
+    - 'Cho mình xem bảng size áo nam.'
+    - 'Phí vận chuyển đi tỉnh là bao nhiêu?'
+    - 'Canifa thành lập năm nào?'
+    """
+    logger.info(f"🔍 [Semantic Search] Brand Knowledge query: {query}")
+
+    try:
+        # 1. Tạo embedding cho câu hỏi (Mặc định 1536 chiều như bro yêu cầu)
+        query_vector = await create_embedding_async(query)
+        if not query_vector:
+            return "Xin lỗi, tôi gặp sự cố khi xử lý thông tin. Vui lòng thử lại sau."
+
+        v_str = "[" + ",".join(str(v) for v in query_vector) + "]"
+
+        # 2. Query StarRocks lấy Top 4 kết quả phù hợp nhất (Không check score)
+        sql = f"""
+        SELECT 
+            content,
+            metadata
+        FROM shared_source.chatbot_rsa_knowledge
+        ORDER BY approx_cosine_similarity(embedding, {v_str}) DESC
+        LIMIT 4
+        """
+
+        sr = get_db_connection()
+        results = await sr.execute_query_async(sql)
+
+        if not results:
+            logger.warning(f"⚠️ No knowledge data found in DB for query: {query}")
+            return "Hiện tại tôi chưa tìm thấy thông tin chính xác về nội dung này trong hệ thống kiến thức của Canifa. Bạn có thể liên hệ hotline 1800 6061 để được hỗ trợ trực tiếp."
+
+        # 3. Tổng hợp kết quả
+        knowledge_texts = []
+        for i, res in enumerate(results):
+            content = res.get("content", "")
+            knowledge_texts.append(content)
+            # LOG DỮ LIỆU LẤY ĐƯỢC (Chỉ hiển thị nội dung)
+            logger.info(f"📄 [Knowledge Chunk {i + 1}]: {content[:200]}...")
+
+        final_response = "\n\n---\n\n".join(knowledge_texts)
+        logger.info(f"✅ Found {len(results)} relevant knowledge chunks.")
+
+        return final_response
+
+    except Exception as e:
+        logger.error(f"❌ Error in canifa_knowledge_search: {e}")
+        return "Tôi đang gặp khó khăn khi truy cập kho kiến thức. Bạn muốn hỏi về sản phẩm gì khác không?"

backend/agent/tools/customer_info_tool.py

+"""
+Tool thu thập thông tin khách hàng (Tên, Số điện thoại, Email)
+Dùng để đẩy data về CRM hoặc hệ thống lưu trữ khách hàng.
+"""
+
+import json
+import logging
+
+from langchain_core.tools import tool
+
+logger = logging.getLogger(__name__)
+
+
+@tool
+async def collect_customer_info(name: str, phone: str, email: str | None) -> str:
+    """
+    Sử dụng tool này để ghi lại thông tin khách hàng khi họ muốn tư vấn sâu hơn,
+    nhận khuyến mãi hoặc đăng ký mua hàng.
+    Args:
+        name: Tên của khách hàng
+        phone: Số điện thoại của khách hàng
+        email: Email của khách hàng (không bắt buộc)
+    """
+    try:
+        print(f"\n[TOOL] --- 📝 Thu thập thông tin khách hàng: {name} - {phone} ---")
+        logger.info(f"📝 Collecting customer info: {name}, {phone}, {email}")
+
+        # Giả lập việc đẩy data đi (CRM/Sheet)
+        # Trong thực tế, bạn sẽ gọi một API ở đây
+        db_record = {
+            "customer_name": name,
+            "phone_number": phone,
+            "email_address": email,
+            "status": "pending_consultation",
+        }
+
+        # Trả về kết quả thành công
+        return json.dumps(
+            {
+                "status": "success",
+                "message": (
+                    f"Cảm ơn anh/chị {name}. CiCi đã ghi nhận thông tin và sẽ có nhân viên "
+                    f"liên hệ tư vấn qua số điện thoại {phone} sớm nhất ạ!"
+                ),
+                "data_captured": db_record,
+            },
+            ensure_ascii=False,
+        )
+
+    except Exception as e:
+        logger.error(f"❌ Lỗi khi thu thập thông tin: {e}")
+        return json.dumps(
+            {
+                "status": "error",
+                "message": f"Xin lỗi, CiCi gặp sự cố khi lưu thông tin. Anh/chị vui lòng thử lại sau ạ. Lỗi: {e!s}",
+            },
+            ensure_ascii=False,
+        )

backend/agent/tools/data_retrieval_tool.py

+"""
+CANIFA Data Retrieval Tool - Tối giản cho Agentic Workflow.
+Hỗ trợ Hybrid Search: Semantic (Vector) + Metadata Filter.
+"""
+
+import asyncio
+import json
+import logging
+import time
+from decimal import Decimal
+
+from langchain_core.tools import tool
+from pydantic import BaseModel, Field
+
+from agent.tools.product_search_helpers import build_starrocks_query
+from common.embedding_service import create_embeddings_async
+from common.starrocks_connection import get_db_connection
+
+# from langsmith import traceable
+
+logger = logging.getLogger(__name__)
+
+
+class DecimalEncoder(json.JSONEncoder):
+    """Xử lý kiểu Decimal từ Database khi convert sang JSON."""
+
+    def default(self, obj):
+        if isinstance(obj, Decimal):
+            return float(obj)
+        return super().default(obj)
+
+
+class SearchItem(BaseModel):
+    """
+    Cấu trúc một mục tìm kiếm đơn lẻ trong Multi-Search.
+
+    Lưu ý quan trọng về cách SINH QUERY:
+    - Trường `query` KHÔNG phải câu hỏi thô của khách.
+    - Phải là một đoạn text có cấu trúc giống hệt format trong cột `description_text_full` của DB,
+      ví dụ (chỉ là 1 chuỗi duy nhất, nối các field bằng dấu chấm):
+
+      product_name: Pack 3 đôi tất bé gái cổ thấp. master_color: Xanh da trời/ Blue.
+      product_image_url: https://.... product_image_url_thumbnail: https://....
+      product_web_url: https://.... description_text: ... material: ...
+      material_group: Yarn - Sợi. gender_by_product: female. age_by_product: others.
+      season: Year. style: Feminine. fitting: Slim. size_scale: 4/6.
+      form_neckline: None. form_sleeve: None. product_line_vn: Tất.
+      product_color_name: Blue Strip 449.
+
+    - Khi khách chỉ nói “áo màu hồng”, hãy suy luận và sinh query dạng:
+
+      product_name: Áo thun/áo sơ mi/áo ... màu hồng ... . master_color: Hồng/ Pink.
+      product_image_url: None. product_image_url_thumbnail: None.
+      product_web_url: None. description_text: ... (mô tả thêm nếu có).
+      material: None. material_group: None. gender_by_product: ... (nếu đoán được).
+      age_by_product: others. season: Year. style: ... (nếu đoán được).
+      fitting: ... size_scale: None. form_neckline: None. form_sleeve: None.
+      product_line_vn: Áo. product_color_name: Pink / Hồng (nếu hợp lý).
+
+    - Nếu không suy luận được giá trị cho field nào thì để `None` hoặc bỏ trống phần text đó.
+    """
+
+    query: str = Field(
+        ...,
+        description=(
+            "ĐOẠN TEXT CÓ CẤU TRÚC theo format của cột description_text_full trong DB, "
+            "bao gồm các cặp key: product_name, master_color, product_image_url, "
+            "product_image_url_thumbnail, product_web_url, description_text, material, "
+            "material_group, gender_by_product, age_by_product, season, style, fitting, "
+            "size_scale, form_neckline, form_sleeve, product_line_vn, product_color_name. "
+            "Ví dụ: 'product_name: Pack 3 đôi tất bé gái cổ thấp. master_color: Xanh da trời/ Blue. "
+            "product_image_url: https://.... product_web_url: https://.... description_text: ... "
+            "material: None. material_group: Yarn - Sợi. gender_by_product: female. ...'"
+        ),
+    )
+    magento_ref_code: str | None = Field(
+        ..., description="Mã sản phẩm hoặc SKU (Ví dụ: 8TS24W001). CHỈ điền khi khách hỏi mã code cụ thể."
+    )
+    price_min: float | None = Field(..., description="Giá thấp nhất (VD: 100000)")
+    price_max: float | None = Field(..., description="Giá cao nhất (VD: 500000)")
+    action: str = Field(..., description="Hành động: 'search' (tìm kiếm) hoặc 'visual_search' (phân tích ảnh)")
+
+
+class MultiSearchParams(BaseModel):
+    """Tham số cho Parallel Multi-Search."""
+
+    searches: list[SearchItem] = Field(..., description="Danh sách các truy vấn tìm kiếm chạy song song")
+
+
+
+@tool(args_schema=MultiSearchParams)
+# @traceable(run_type="tool", name="data_retrieval_tool")
+async def data_retrieval_tool(searches: list[SearchItem]) -> str:
+    """
+    Siêu công cụ tìm kiếm sản phẩm CANIFA - Hỗ trợ Parallel Multi-Search (chạy song song nhiều truy vấn).
+
+    Hướng dẫn dùng nhanh:
+    - Trường 'query': mô tả chi tiết sản phẩm (tên, chất liệu, giới tính, màu sắc, phong cách, dịp sử dụng), không dùng câu hỏi thô.
+    - Trường 'magento_ref_code': chỉ dùng khi khách hỏi mã sản phẩm/SKU cụ thể (vd: 8TS24W001).
+    - Trường 'price_min' / 'price_max': dùng khi khách nói về khoảng giá (vd: dưới 500k, từ 200k đến 400k).
+    """
+    logger.info("data_retrieval_tool started, searches=%s", len(searches))
+    try:
+        # 0. Log input tổng quan (không log chi tiết dài)
+        for idx, item in enumerate(searches):
+            short_query = (item.query[:60] + "...") if item.query and len(item.query) > 60 else item.query
+            logger.debug(
+                "search[%s] query=%r, code=%r, price_min=%r, price_max=%r",
+                idx,
+                short_query,
+                item.magento_ref_code,
+                item.price_min,
+                item.price_max,
+            )
+
+
+        queries_to_embed = [s.query for s in searches if s.query]
+        all_vectors = []
+        if queries_to_embed:
+            logger.info("batch embedding %s queries", len(queries_to_embed))
+            emb_batch_start = time.time()
+            all_vectors = await create_embeddings_async(queries_to_embed)
+            logger.info(
+                "batch embedding done in %.2f ms",
+                (time.time() - emb_batch_start) * 1000,
+            )
+
+        # 2. Get DB connection (singleton)
+        db = get_db_connection()
+
+        tasks = []
+        vector_idx = 0
+        for item in searches:
+            current_vector = None
+            if item.query:
+                if vector_idx < len(all_vectors):
+                    current_vector = all_vectors[vector_idx]
+                    vector_idx += 1
+
+            tasks.append(_execute_single_search(db, item, query_vector=current_vector))
+
+        results = await asyncio.gather(*tasks)
+
+        # 3. Tổng hợp kết quả
+        combined_results = []
+        for i, products in enumerate(results):
+            combined_results.append(
+                {
+                    "search_index": i,
+                    "search_criteria": searches[i].dict(exclude_none=True),
+                    "count": len(products),
+                    "products": products,
+                }
+            )
+
+        logger.info("data_retrieval_tool finished, results=%s", len(combined_results))
+        return json.dumps(
+            {"status": "success", "results": combined_results},
+            ensure_ascii=False,
+            cls=DecimalEncoder,
+        )
+
+    except Exception as e:
+        logger.exception("Error in Multi-Search data_retrieval_tool: %s", e)
+        return json.dumps({"status": "error", "message": str(e)})
+
+
+async def _execute_single_search(db, item: SearchItem, query_vector: list[float] | None = None) -> list[dict]:
+    """Thực thi một search query đơn lẻ (Async)."""
+    try:
+        short_query = (item.query[:60] + "...") if item.query and len(item.query) > 60 else item.query
+        logger.debug(
+            "_execute_single_search started, query=%r, code=%r",
+            short_query,
+            item.magento_ref_code,
+        )
+
+        # Timer: build query (sử dụng vector đã có hoặc build mới)
+        query_build_start = time.time()
+        sql = await build_starrocks_query(item, query_vector=query_vector)
+        query_build_time = (time.time() - query_build_start) * 1000  # Convert to ms
+        logger.debug("SQL built, length=%s, build_time_ms=%.2f", len(sql), query_build_time)
+
+        # Timer: execute DB query
+        db_start = time.time()
+        products = await db.execute_query_async(sql)
+        db_time = (time.time() - db_start) * 1000  # Convert to ms
+        logger.info(
+            "_execute_single_search done, products=%s, build_ms=%.2f, db_ms=%.2f, total_ms=%.2f",
+            len(products),
+            query_build_time,
+            db_time,
+            query_build_time + db_time,
+        )
+
+        return _format_product_results(products)
+    except Exception as e:
+        logger.exception("Single search error for item %r: %s", item, e)
+        return []
+
+
+def _format_product_results(products: list[dict]) -> list[dict]:
+    """Lọc và format kết quả trả về cho Agent."""
+    max_items = 15
+    formatted: list[dict] = []
+
+    for p in products[:max_items]:
+        formatted.append(
+            {
+                "internal_ref_code": p.get("internal_ref_code"),
+                # Chuỗi text dài, đã bao gồm: product_name, master_color, image, web_url, material, style, ...
+                "description_text": p.get("description_text_full"),
+                "sale_price": p.get("sale_price"),
+                "original_price": p.get("original_price"),
+                "discount_amount": p.get("discount_amount"),
+                "max_score": p.get("max_score"),
+            }
+        )
+
+    return formatted

backend/agent/tools/get_tools.py

+"""
+Tools Factory
+Chỉ return 1 tool duy nhất: data_retrieval_tool
+"""
+
+from langchain_core.tools import Tool
+
+from .brand_knowledge_tool import canifa_knowledge_search
+from .customer_info_tool import collect_customer_info
+from .data_retrieval_tool import data_retrieval_tool
+
+
+def get_retrieval_tools() -> list[Tool]:
+    """Các tool chỉ dùng để đọc/truy vấn dữ liệu (Có thể cache)"""
+    return [data_retrieval_tool, canifa_knowledge_search]
+
+
+def get_collection_tools() -> list[Tool]:
+    """Các tool dùng để ghi/thu thập dữ liệu (KHÔNG cache)"""
+    return [collect_customer_info]
+
+
+def get_all_tools() -> list[Tool]:
+    """Return toàn bộ list tools cho Agent"""
+    return get_retrieval_tools() + get_collection_tools()

backend/agent/tools/product_search_helpers.py

+import logging
+import time
+
+from common.embedding_service import create_embedding_async
+
+logger = logging.getLogger(__name__)
+
+
+def _escape(val: str) -> str:
+    """Thoát dấu nháy đơn để tránh SQL Injection cơ bản."""
+    return val.replace("'", "''")
+
+
+def _get_where_clauses(params) -> list[str]:
+    """Xây dựng danh sách các điều kiện lọc từ params."""
+    clauses = []
+    clauses.extend(_get_price_clauses(params))
+    clauses.extend(_get_metadata_clauses(params))
+    clauses.extend(_get_special_clauses(params))
+    return clauses
+
+
+def _get_price_clauses(params) -> list[str]:
+    """Lọc theo giá."""
+    clauses = []
+    p_min = getattr(params, "price_min", None)
+    if p_min is not None:
+        clauses.append(f"sale_price >= {p_min}")
+    p_max = getattr(params, "price_max", None)
+    if p_max is not None:
+        clauses.append(f"sale_price <= {p_max}")
+    return clauses
+
+
+def _get_metadata_clauses(params) -> list[str]:
+    """Xây dựng điều kiện lọc từ metadata (Phối hợp Exact và Partial)."""
+    clauses = []
+
+    # 1. Exact Match (Giới tính, Độ tuổi) - Các trường này cần độ chính xác tuyệt đối
+    exact_fields = [
+        ("gender_by_product", "gender_by_product"),
+        ("age_by_product", "age_by_product"),
+    ]
+    for param_name, col_name in exact_fields:
+        val = getattr(params, param_name, None)
+        if val:
+            clauses.append(f"{col_name} = '{_escape(val)}'")
+
+    # 2. Partial Match (LIKE) - Giúp map text linh hoạt hơn (Chất liệu, Dòng SP, Phong cách...)
+    # Cái này giúp map: "Yarn" -> "Yarn - Sợi", "Knit" -> "Knit - Dệt Kim"
+    partial_fields = [
+        ("season", "season"),
+        ("material_group", "material_group"),
+        ("product_line_vn", "product_line_vn"),
+        ("style", "style"),
+        ("fitting", "fitting"),
+        ("form_neckline", "form_neckline"),
+        ("form_sleeve", "form_sleeve"),
+    ]
+    for param_name, col_name in partial_fields:
+        val = getattr(params, param_name, None)
+        if val:
+            v = _escape(val).lower()
+            # Dùng LOWER + LIKE để cân mọi loại ký tự thừa hoặc hoa/thường
+            clauses.append(f"LOWER({col_name}) LIKE '%{v}%'")
+
+    return clauses
+
+
+def _get_special_clauses(params) -> list[str]:
+    """Các trường hợp đặc biệt: Mã sản phẩm, Màu sắc."""
+    clauses = []
+    # Mã sản phẩm / SKU
+    m_code = getattr(params, "magento_ref_code", None)
+    if m_code:
+        m = _escape(m_code)
+        clauses.append(f"(magento_ref_code = '{m}' OR internal_ref_code = '{m}')")
+
+    # Màu sắc
+    color = getattr(params, "master_color", None)
+    if color:
+        c = _escape(color).lower()
+        clauses.append(f"(LOWER(master_color) LIKE '%{c}%' OR LOWER(product_color_name) LIKE '%{c}%')")
+    return clauses
+
+
+async def build_starrocks_query(params, query_vector: list[float] | None = None) -> str:
+    """
+    Build SQL cho Product Search với 2 chiến lược:
+    1. CODE SEARCH: Nếu có magento_ref_code → Tìm trực tiếp theo mã (KHÔNG dùng vector)
+    2. HYDE SEARCH: Semantic search với HyDE vector (Pure vector approach)
+    """
+
+    # ============================================================
+    # CASE 1: CODE SEARCH - Tìm theo mã sản phẩm (No Vector)
+    # ============================================================
+    magento_code = getattr(params, "magento_ref_code", None)
+    if magento_code:
+        logger.info(f"🎯 [CODE SEARCH] Direct search by code: {magento_code}")
+        code = _escape(magento_code)
+
+        # Tìm trực tiếp theo mã + Lọc trùng (GROUP BY internal_ref_code)
+        # Tìm chính xác theo mã (Lấy tất cả các bản ghi/màu sắc/size của mã đó)
+        sql = f"""
+        SELECT 
+            internal_ref_code,
+            description_text_full,
+            sale_price,
+            original_price,
+            discount_amount,
+            1.0 as max_score
+        FROM shared_source.magento_product_dimension_with_text_embedding
+        WHERE (magento_ref_code = '{code}' OR internal_ref_code = '{code}')
+        """
+
+        print("✅ [CODE SEARCH] Query built - No vector search needed!")
+
+        # Ghi log debug query FULL vào Background Task (Không làm chậm Request)
+        # asyncio.create_task(save_query_to_log(sql))
+
+        return sql
+
+    # ============================================================
+    # CASE 2: HYDE SEARCH - Semantic Vector Search
+    # ============================================================
+    logger.info("🚀 [HYDE RETRIEVER] Starting semantic vector search...")
+
+    # 1. Lấy Vector từ HyDE (AI-generated hypothetical document)
+    query_text = getattr(params, "query", None)
+
+    if query_text and query_vector is None:
+        emb_start = time.time()
+        query_vector = await create_embedding_async(query_text)
+        logger.info(f"⏱️ [TIMER] Single HyDE Embedding: {(time.time() - emb_start) * 1000:.2f}ms")
+
+    if not query_vector:
+        logger.warning("⚠️ No vector found, returning empty query.")
+        return ""
+
+    v_str = "[" + ",".join(str(v) for v in query_vector) + "]"
+
+    # 2. Build PRICE filter ONLY (chỉ lọc giá, để vector tự semantic search)
+    price_clauses = _get_price_clauses(params)
+    where_filter = ""
+    if price_clauses:
+        where_filter = " AND " + " AND ".join(price_clauses)
+        logger.info(f"💰 [PRICE FILTER] Applied: {where_filter}")
+
+    # 3. SQL Pure Vector Search + Price Filter Only
+    sql = f"""
+    WITH top_matches AS (
+        SELECT /*+ SET_VAR(ann_params='{{"ef_search":128}}') */
+            internal_ref_code,
+            product_color_code,
+            description_text_full,
+            sale_price,
+            original_price,
+            discount_amount,
+            approx_cosine_similarity(vector, {v_str}) as similarity_score
+        FROM shared_source.magento_product_dimension_with_text_embedding
+        ORDER BY similarity_score DESC
+        LIMIT 100
+    )
+    SELECT 
+        internal_ref_code,
+        MAX_BY(description_text_full, similarity_score) as description_text_full,
+        MAX_BY(sale_price, similarity_score) as sale_price,
+        MAX_BY(original_price, similarity_score) as original_price,
+        MAX_BY(discount_amount, similarity_score) as discount_amount,
+        MAX(similarity_score) as max_score
+    FROM top_matches
+    WHERE 1=1 {where_filter}
+    GROUP BY internal_ref_code
+    ORDER BY max_score DESC
+    LIMIT 20
+    """
+
+    return sql
+
+
+# ============================================================
+# TEMPORARILY COMMENTED OUT - save_query_to_log
+# ============================================================
+# async def save_query_to_log(sql: str):
+#     """Lưu query full vào file hyde_pure_query.txt."""
+#     import os
+#     log_path = r"D:\cnf\chatbot_canifa\backend\logs\hyde_pure_query.txt"
+#     try:
+#         log_dir = os.path.dirname(log_path)
+#         if not os.path.exists(log_dir):
+#             os.makedirs(log_dir)
+#         with open(log_path, "w", encoding="utf-8") as f:
+#             f.write(sql)
+#         print(f"💾 Full Query saved to: {log_path}")
+#     except Exception as e:
+#         print(f"Save query log failed: {e}")
+
+
+# ============================================================
+# TEMPORARILY COMMENTED OUT - save_preview_to_log
+# ============================================================
+# async def save_preview_to_log(search_query: str, products: list[dict]):
+#     """Lưu kết quả DB trả về vào db_preview.txt (Format đẹp cho AI)."""
+#     import os
+#     preview_path = r"D:\cnf\chatbot_canifa\backend\logs\db_preview.txt"
+#     try:
+#         log_dir = os.path.dirname(preview_path)
+#         if not os.path.exists(log_dir):
+#             os.makedirs(log_dir)
+#
+#         with open(preview_path, "a", encoding="utf-8") as f:
+#             f.write(f"\n{'='*60}\n")
+#             f.write(f"⏰ TIME: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
+#             f.write(f"🔍 SEARCH: {search_query}\n")
+#             f.write(f"📊 RESULTS COUNT: {len(products)}\n")
+#             f.write(f"{'-'*60}\n")
+#
+#             if not products:
+#                 f.write("❌ NO PRODUCTS FOUND\n")
+#             else:
+#                 for idx, p in enumerate(products[:5], 1):
+#                     code = p.get("internal_ref_code", "N/A")
+#                     sale = p.get("sale_price", "N/A")
+#                     orig = p.get("original_price", "N/A")
+#                     disc = p.get("discount_amount", "0")
+#                     score = p.get("max_score", p.get("similarity_score", "N/A"))
+#                     desc = p.get("description_text_full", "No Description")
+#
+#                     f.write(f"{idx}. [{code}] Score: {score}\n")
+#                     f.write(f"   💰 Price: {sale} (Orig: {orig}, Disc: {disc}%)\n")
+#                     f.write(f"   📝 Desc: {desc}\n")
+#
+#             f.write(f"{'='*60}\n")
+#         print(f"💾 DB Preview (Results) saved to: {preview_path}")
+#     except Exception as e:
+#         print(f"Save preview log failed: {e}")

backend/api/cache_analytics_route.py

+"""
+Cache Analytics API Routes
+===========================
+
+Provides endpoints to monitor semantic cache performance:
+- Cache statistics (hit rate, cost savings, performance)
+- Clear user cache
+- Reset statistics
+"""
+
+import logging
+
+from fastapi import APIRouter
+
+from common.cache import clear_user_cache, get_cache_stats, reset_cache_stats
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/cache", tags=["Cache Analytics"])
+
+
+@router.get("/stats")
+async def get_cache_statistics():
+    """
+    Get semantic cache performance statistics.
+    
+    Returns:
+        Cache stats including:
+        - LLM cache hit/miss rates
+        - Embedding cache hit/miss rates  
+        - Cost savings (USD)
+        - Performance metrics (time saved)
+    
+    Example Response:
+    ```json
+    {
+        "total_queries": 150,
+        "llm_cache": {
+            "hits": 90,
+            "misses": 60,
+            "hit_rate_percent": 60.0,
+            "cost_saved_usd": 0.09
+        },
+        "embedding_cache": {
+            "hits": 120,
+            "misses": 30,
+            "hit_rate_percent": 80.0,
+            "cost_saved_usd": 0.012
+        },
+        "performance": {
+            "avg_saved_time_ms": 1850,
+            "total_time_saved_seconds": 166.5
+        },
+        "total_cost_saved_usd": 0.102
+    }
+    ```
+    """
+    try:
+        stats = await get_cache_stats()
+        return {
+            "status": "success",
+            "data": stats,
+        }
+    except Exception as e:
+        logger.error(f"Error getting cache stats: {e}", exc_info=True)
+        return {
+            "status": "error",
+            "message": str(e),
+        }
+
+
+@router.delete("/user/{user_id}")
+async def clear_cache_for_user(user_id: str):
+    """
+    Clear all cached responses for a specific user.
+    
+    Args:
+        user_id: User ID to clear cache for
+    
+    Returns:
+        Number of cache entries deleted
+    
+    Use cases:
+    - User requests to clear their data
+    - User reports incorrect cached responses
+    - Manual cache invalidation for testing
+    """
+    try:
+        deleted_count = await clear_user_cache(user_id)
+        return {
+            "status": "success",
+            "message": f"Cleared {deleted_count} cache entries for user {user_id}",
+            "deleted_count": deleted_count,
+        }
+    except Exception as e:
+        logger.error(f"Error clearing user cache: {e}", exc_info=True)
+        return {
+            "status": "error",
+            "message": str(e),
+        }
+
+
+@router.post("/stats/reset")
+async def reset_statistics():
+    """
+    Reset cache statistics counters.
+    
+    This resets:
+    - Hit/miss counters
+    - Cost savings calculations
+    - Performance metrics
+    
+    Note: This does NOT delete cached data, only resets the statistics.
+    """
+    try:
+        reset_cache_stats()
+        return {
+            "status": "success",
+            "message": "Cache statistics reset successfully",
+        }
+    except Exception as e:
+        logger.error(f"Error resetting cache stats: {e}", exc_info=True)
+        return {
+            "status": "error",
+            "message": str(e),
+        }

backend/api/chatbot_route.py

+"""
+Fashion Q&A Agent Router
+FastAPI endpoints cho Fashion Q&A Agent service.
+Router chỉ chứa định nghĩa API, logic nằm ở controller.
+
+Note: Rate limit check đã được xử lý trong middleware (CanifaAuthMiddleware)
+"""
+
+import logging
+
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
+from opentelemetry import trace
+
+from agent.controller import chat_controller
+from agent.models import QueryRequest
+from common.message_limit import message_limit_service
+from common.user_identity import get_user_identity
+from config import DEFAULT_MODEL
+
+logger = logging.getLogger(__name__)
+tracer = trace.get_tracer(__name__)
+router = APIRouter()
+
+
+@router.post("/api/agent/chat", summary="Fashion Q&A Chat (Non-streaming)")
+async def fashion_qa_chat(request: Request, req: QueryRequest, background_tasks: BackgroundTasks):
+    """
+    Endpoint chat không stream - trả về response JSON đầy đủ một lần.
+    
+    Note: Rate limit đã được check trong middleware.
+    """
+    # 1. Xác định user identity
+    identity = get_user_identity(request)
+    user_id = identity.primary_id
+    
+    # Rate limit đã check trong middleware, lấy limit_info từ request.state
+    limit_info = getattr(request.state, 'limit_info', None)
+
+    logger.info(f"📥 [Incoming Query - NonStream] User: {user_id} | Query: {req.user_query}")
+
+    # Get current span để add logs VÀO JAEGER UI
+    span = trace.get_current_span()
+    span.set_attribute("user.id", user_id)
+    span.set_attribute("chat.user_query", req.user_query)
+    span.add_event(
+        "📥 User query received", attributes={"user_id": user_id, "query": req.user_query, "timestamp": "incoming"}
+    )
+
+    try:
+        # Gọi controller để xử lý logic (Non-streaming)
+        result = await chat_controller(
+            query=req.user_query,
+            user_id=user_id,
+            background_tasks=background_tasks,
+            model_name=DEFAULT_MODEL,
+            images=req.images,
+            identity_key=identity.history_key,  # Guest: device_id, User: user_id
+        )
+
+        # Log chi tiết response
+        logger.info(f"📤 [Outgoing Response - NonStream] User: {user_id}")
+        logger.info(f"💬 AI Response: {result['ai_response']}")
+        logger.info(f"🛍️ Product IDs: {result.get('product_ids', [])}")
+
+        # Add to span (hiển thị trong Jaeger UI)
+        span.set_attribute("chat.ai_response", result["ai_response"][:200])  # Giới hạn 200 ký tự
+        span.set_attribute("chat.product_count", len(result.get("product_ids", [])))
+        span.add_event(
+            "💬 AI response generated",
+            attributes={
+                "ai_response_preview": result["ai_response"][:100],
+                "product_count": len(result.get("product_ids", [])),
+                "product_ids": str(result.get("product_ids", [])[:5]),  # First 5 IDs
+            },
+        )
+
+        # Increment message count SAU KHI chat thành công
+        usage_info = await message_limit_service.increment(
+            identity_key=identity.rate_limit_key,
+            is_authenticated=identity.is_authenticated,
+        )
+
+        return {
+            "status": "success",
+            "ai_response": result["ai_response"],
+            "product_ids": result.get("product_ids", []),
+            "limit_info": {
+                "limit": usage_info["limit"],
+                "used": usage_info["used"],
+                "remaining": usage_info["remaining"],
+            },
+        }
+    except Exception as e:
+        logger.error(f"Error in fashion_qa_chat: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e)) from e
+

backend/api/conservation_route.py

+"""
+Chat History API Routes
+- GET /api/history/{identity_key} - Lấy lịch sử chat (có product_ids)
+- DELETE /api/history/{identity_key} - Xóa lịch sử chat
+
+Note: identity_key có thể là device_id (guest) hoặc user_id (đã login)
+"""
+
+import logging
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+from common.conversation_manager import get_conversation_manager
+from common.user_identity import get_user_identity
+
+router = APIRouter(tags=["Chat History"])
+logger = logging.getLogger(__name__)
+
+
+class ChatHistoryResponse(BaseModel):
+    data: list[dict[str, Any]]
+    next_cursor: int | None = None
+
+
+class ClearHistoryResponse(BaseModel):
+    success: bool
+    message: str
+
+
+@router.get("/api/history/{identity_key}", summary="Get Chat History", response_model=ChatHistoryResponse)
+async def get_chat_history(request: Request, identity_key: str, limit: int | None = 50, before_id: int | None = None):
+    """
+    Lấy lịch sử chat theo identity_key.
+    
+    Logic: Middleware đã parse token -> Nếu user đã login thì dùng user_id, không thì dùng device_id.
+    (identity_key trong URL chỉ là fallback)
+    """
+    try:
+        # Tự động resolve identity từ middleware
+        identity = get_user_identity(request)
+        resolved_key = identity.history_key  # user_id nếu login, device_id nếu không
+        
+        logger.info(f"GET History: URL key={identity_key} -> Resolved key={resolved_key}")
+        
+        manager = await get_conversation_manager()
+        history = await manager.get_chat_history(resolved_key, limit=limit, before_id=before_id)
+
+        next_cursor = None
+        if history and len(history) > 0:
+            next_cursor = history[-1]["id"]
+
+        return {"data": history, "next_cursor": next_cursor}
+    except Exception as e:
+        logger.error(f"Error fetching chat history: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch chat history")
+
+
+@router.delete("/api/history/{identity_key}", summary="Clear Chat History", response_model=ClearHistoryResponse)
+async def clear_chat_history(identity_key: str):
+    """
+    Xóa toàn bộ lịch sử chat theo identity_key. 
+    Logic: Middleware đã parse token -> Nếu user đã login thì dùng user_id, không thì dùng device_id.
+    """
+    try:
+        manager = await get_conversation_manager()
+        await manager.clear_history(identity_key)
+        logger.info(f"✅ Cleared chat history for {identity_key}")
+        return {"success": True, "message": f"Đã xóa lịch sử chat của {identity_key}"}
+    except Exception as e:
+        logger.error(f"Error clearing chat history for {identity_key}: {e}")
+        raise HTTPException(status_code=500, detail="Failed to clear chat history")
+

backend/api/memos/__init__.py

+from fastapi import APIRouter
+
+from .instance_routes import router as instance_router
+from .auth_routes import router as auth_router
+from .user_routes import router as user_router
+from .memo_routes import router as memo_router
+from .attachment_routes import router as attachment_router
+from .shortcut_routes import router as shortcut_router
+from .activity_routes import router as activity_router
+from .idp_routes import router as idp_router
+
+
+router = APIRouter(prefix="/api/v1")
+
+
+router.include_router(instance_router, tags=["instance"])
+router.include_router(auth_router, tags=["auth"])
+router.include_router(user_router, tags=["users"])
+router.include_router(memo_router, tags=["memos"])
+router.include_router(attachment_router, tags=["attachments"])
+router.include_router(shortcut_router, tags=["shortcuts"])
+router.include_router(activity_router, tags=["activities"])
+router.include_router(idp_router, tags=["idp"])
+
+

backend/api/memos/activity_routes.py

+"""
+Activity service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from memos_core.schemas import ActivityResponse
+from memos_core.services import get_activity_service
+
+
+router = APIRouter(prefix="/activities")
+
+
+@router.get("", summary="List activities", response_model=List[ActivityResponse])
+async def list_activities(activity_service=Depends(get_activity_service)):
+    try:
+        return await activity_service.list_activities()
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+

backend/api/memos/attachment_routes.py

+"""
+Attachment service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
+
+from memos_core.schemas import (
+    AttachmentResponse,
+)
+from memos_core.services import get_attachment_service
+
+
+router = APIRouter(prefix="/attachments")
+
+
+@router.get("", summary="List attachments", response_model=List[AttachmentResponse])
+async def list_attachments(attachment_service=Depends(get_attachment_service)):
+    try:
+        return await attachment_service.list_attachments()
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="Upload attachment", response_model=AttachmentResponse)
+async def upload_attachment(
+    file: UploadFile = File(...),
+    attachment_service=Depends(get_attachment_service),
+):
+    try:
+        return await attachment_service.upload_attachment(file)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/{attachment_id}", summary="Get attachment by ID", response_model=AttachmentResponse)
+async def get_attachment(attachment_id: int, attachment_service=Depends(get_attachment_service)):
+    try:
+        return await attachment_service.get_attachment(attachment_id)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+

backend/api/memos/auth_routes.py

+"""
+Auth service routes for Memos-style backend.
+
+Router chỉ chứa định nghĩa API, logic xác thực nằm ở memos_core.services.auth.
+"""
+
+# ruff: noqa: I001
+from fastapi import APIRouter, Body, Depends, HTTPException
+
+from memos_core.schemas import (
+    AuthSignInRequest,
+    AuthSignInResponse,
+    AuthSignUpRequest,
+    AuthSignUpResponse,
+)
+from memos_core.services import get_auth_service
+
+
+router = APIRouter(prefix="/auth")
+
+
+@router.post("/signin", summary="Sign in", response_model=AuthSignInResponse)
+async def signin(
+    payload: dict = Body(default_factory=dict),  # noqa: B008
+    auth_service=Depends(get_auth_service),  # noqa: B008 (FastAPI dependency)
+):
+    try:
+        # Accept both REST payload and Connect/proto-like envelope.
+        # REST: { "email": "...", "password": "..." }
+        # Connect/proto-like: { "credentials": { "case": "passwordCredentials", "value": { "username": "...", "password": "..." } } }
+        raw = payload if isinstance(payload, dict) else {}
+
+        # Normalize Connect-style password credentials into legacy fields.
+        creds = raw.get("credentials") if isinstance(raw, dict) else None
+        if isinstance(creds, dict):
+            case = creds.get("case")
+            value = creds.get("value")
+            if case == "passwordCredentials" and isinstance(value, dict):
+                raw = {
+                    "email": value.get("username"),
+                    "password": value.get("password"),
+                    **raw,
+                }
+
+        signin_req = AuthSignInRequest(**(raw or {}))
+        result = await auth_service.sign_in(signin_req)
+        # Ensure camelCase keys for Connect/proto TS client.
+        return result.model_dump(by_alias=True)
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.post("/signup", summary="Sign up", response_model=AuthSignUpResponse)
+async def signup(
+    payload: AuthSignUpRequest,
+    auth_service=Depends(get_auth_service),  # noqa: B008 (FastAPI dependency)
+):
+    try:
+        return await auth_service.sign_up(payload)
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/me", summary="Get current user")
+@router.post("/me", summary="Get current user (Connect compatibility)")
+async def get_me(auth_service=Depends(get_auth_service)):  # noqa: B008 (FastAPI dependency)
+    try:
+        me = await auth_service.get_me()
+        # Connect/proto expects: { "user": User }
+        return {
+            "user": {
+                "name": "users/1",
+                "role": 1,  # HOST
+                "username": me.email.split("@", 1)[0] if getattr(me, "email", None) else "demo",
+                "email": getattr(me, "email", "demo@example.com"),
+                "displayName": "",
+                "avatarUrl": "",
+                "description": "",
+                "password": "",
+                "state": 1,
+            }
+        }
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=401, detail=str(exc)) from exc
+
+

backend/api/memos/idp_routes.py

+"""
+Identity Provider (IDP) service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from memos_core.schemas import (
+    IdentityProviderCreate,
+    IdentityProviderUpdate,
+    IdentityProviderResponse,
+)
+from memos_core.services import get_idp_service
+
+
+router = APIRouter(prefix="/idp")
+
+
+@router.get("", summary="List identity providers", response_model=List[IdentityProviderResponse])
+async def list_identity_providers(idp_service=Depends(get_idp_service)):
+    try:
+        return await idp_service.list_identity_providers()
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="Create identity provider", response_model=IdentityProviderResponse)
+async def create_identity_provider(
+    payload: IdentityProviderCreate,
+    idp_service=Depends(get_idp_service),
+):
+    try:
+        return await idp_service.create_identity_provider(payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.patch("/{idp_id}", summary="Update identity provider", response_model=IdentityProviderResponse)
+async def update_identity_provider(
+    idp_id: int,
+    payload: IdentityProviderUpdate,
+    idp_service=Depends(get_idp_service),
+):
+    try:
+        return await idp_service.update_identity_provider(idp_id, payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+

backend/api/memos/instance_routes.py

+"""
+Instance service routes for Memos-style backend.
+
+Chỉ định nghĩa API, logic sẽ nằm ở memos_core.services.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from memos_core.schemas import InstanceInfo, InstanceSetting, InstanceUpdate
+from memos_core.services import get_instance_service
+
+
+router = APIRouter(prefix="/instance")
+
+
+@router.get("", summary="Get instance info", response_model=InstanceInfo)
+@router.post("", summary="Get instance info (Connect compatibility)", response_model=InstanceInfo)
+async def get_instance(instance_service=Depends(get_instance_service)):
+    try:
+        return await instance_service.get_instance_info()
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.patch("", summary="Update instance basic settings", response_model=InstanceInfo)
+async def update_instance(
+    payload: InstanceUpdate,
+    instance_service=Depends(get_instance_service),
+):
+    try:
+        return await instance_service.update_instance_info(payload)
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("/setting", summary="Get instance setting (Connect compatibility)", response_model=InstanceSetting)
+async def get_instance_setting(instance_service=Depends(get_instance_service)):
+    try:
+        return await instance_service.get_instance_setting()
+    except Exception as exc:  # pragma: no cover - placeholder
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+

backend/api/memos/memo_routes.py

+"""
+Memo service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Body, Depends, HTTPException, Query
+
+from memos_core.schemas import (
+    MemoCreate,
+    MemoUpdate,
+    MemoResponse,
+)
+from memos_core.services import get_memo_service
+
+
+router = APIRouter(prefix="/memos")
+
+
+@router.get("", summary="List memos", response_model=List[MemoResponse])
+async def list_memos(
+    tag: str | None = Query(default=None),
+    memo_service=Depends(get_memo_service),
+):
+    try:
+        return await memo_service.list_memos(tag=tag)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="Create memo (Connect compatibility)")
+async def create_memo_or_list_memos(
+    payload: dict = Body(default_factory=dict),  # noqa: B008
+    memo_service=Depends(get_memo_service),
+):
+    """
+    Vite dev proxy maps multiple Connect RPCs to the same REST path `/api/v1/memos`:
+    - MemoService/ListMemos  -> POST /api/v1/memos  (should be GET but method can't be rewritten)
+    - MemoService/CreateMemo -> POST /api/v1/memos
+
+    To avoid 422 and keep dev UX smooth, accept an untyped payload and branch by shape.
+    """
+    try:
+        raw = payload if isinstance(payload, dict) else {}
+
+        # Connect CreateMemo often wraps payload as { "memo": { ... } }
+        if isinstance(raw.get("memo"), dict):
+            memo_create = MemoCreate(**raw["memo"])
+            return await memo_service.create_memo(memo_create)
+
+        # If it looks like a create payload, also treat as create.
+        if "content" in raw:
+            memo_create = MemoCreate(**raw)
+            return await memo_service.create_memo(memo_create)
+
+        # Otherwise treat as ListMemos.
+        # Support basic tag filter when provided.
+        tag = raw.get("tag") if isinstance(raw.get("tag"), str) else None
+        return await memo_service.list_memos(tag=tag)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/{memo_id}", summary="Get memo by ID", response_model=MemoResponse)
+async def get_memo(memo_id: int, memo_service=Depends(get_memo_service)):
+    try:
+        return await memo_service.get_memo(memo_id)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@router.patch("/{memo_id}", summary="Update memo", response_model=MemoResponse)
+async def update_memo(
+    memo_id: int,
+    payload: MemoUpdate,
+    memo_service=Depends(get_memo_service),
+):
+    try:
+        return await memo_service.update_memo(memo_id, payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.delete("/{memo_id}", summary="Delete memo")
+async def delete_memo(memo_id: int, memo_service=Depends(get_memo_service)):
+    try:
+        await memo_service.delete_memo(memo_id)
+        return {"status": "success"}
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+

backend/api/memos/shortcut_routes.py

+"""
+Shortcut service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Body, Depends, HTTPException
+
+from memos_core.schemas import (
+    ShortcutCreate,
+    ShortcutUpdate,
+    ShortcutResponse,
+    ListShortcutsResponse,
+)
+from memos_core.services import get_shortcut_service
+
+
+router = APIRouter(prefix="/shortcuts")
+
+
+@router.get("", summary="List shortcuts", response_model=List[ShortcutResponse])
+async def list_shortcuts(shortcut_service=Depends(get_shortcut_service)):
+    try:
+        return await shortcut_service.list_shortcuts()
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="List shortcuts (Connect compatibility)", response_model=ListShortcutsResponse)
+async def list_shortcuts_connect_compat(
+    payload: dict = Body(default_factory=dict),  # noqa: B008
+    shortcut_service=Depends(get_shortcut_service),
+):
+    # Connect RPC ListShortcuts is proxied as POST /api/v1/shortcuts in dev.
+    # Ignore payload (parent, pagination...) and return empty list for now.
+    try:
+        _ = payload
+        shortcuts = await shortcut_service.list_shortcuts()
+        return ListShortcutsResponse(shortcuts=shortcuts)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="Create shortcut", response_model=ShortcutResponse)
+async def create_shortcut(
+    payload: ShortcutCreate,
+    shortcut_service=Depends(get_shortcut_service),
+):
+    try:
+        return await shortcut_service.create_shortcut(payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.patch("/{shortcut_id}", summary="Update shortcut", response_model=ShortcutResponse)
+async def update_shortcut(
+    shortcut_id: int,
+    payload: ShortcutUpdate,
+    shortcut_service=Depends(get_shortcut_service),
+):
+    try:
+        return await shortcut_service.update_shortcut(shortcut_id, payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.delete("/{shortcut_id}", summary="Delete shortcut")
+async def delete_shortcut(
+    shortcut_id: int,
+    shortcut_service=Depends(get_shortcut_service),
+):
+    try:
+        await shortcut_service.delete_shortcut(shortcut_id)
+        return {"status": "success"}
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+

backend/api/memos/user_routes.py

+"""
+User service routes for Memos-style backend.
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Body, Depends, HTTPException
+
+from memos_core.schemas import (
+    UserCreate,
+    UserUpdate,
+    UserResponse,
+)
+from memos_core.services import get_user_service
+
+
+router = APIRouter(prefix="/users")
+
+
+@router.get("", summary="List users", response_model=List[UserResponse])
+async def list_users(user_service=Depends(get_user_service)):
+    try:
+        return await user_service.list_users()
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@router.post("", summary="Create user", response_model=UserResponse)
+async def create_user(
+    payload: dict = Body(default_factory=dict),
+    user_service=Depends(get_user_service),
+):
+    try:
+        # Accept both REST payload and Connect/proto-like envelope.
+        # REST: { "email": "...", "password": "..." }
+        # Possible envelope: { "user": { ... } } or other shapes.
+        raw = payload.get("user") if isinstance(payload, dict) else None
+        raw = raw if isinstance(raw, dict) else payload
+        user_create = UserCreate(**(raw or {}))
+        return await user_service.create_user(user_create)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.post("/settings", summary="List user settings (Connect compatibility)")
+async def list_user_settings_connect_compat(payload: dict = Body(default_factory=dict)):  # noqa: B008
+    """
+    Connect RPC UserService/ListUserSettings is proxied as:
+      POST /api/v1/users/settings
+
+    The real Memos API returns a ListUserSettingsResponse with `settings`, `next_page_token`, `total_size`.
+    For stub backend, return empty list to let frontend proceed.
+    """
+    _ = payload
+    return {
+        "settings": [],
+        "next_page_token": "",
+        "total_size": 0,
+    }
+
+
+@router.get("/{user_id}", summary="Get user by ID", response_model=UserResponse)
+async def get_user(user_id: int, user_service=Depends(get_user_service)):
+    try:
+        return await user_service.get_user(user_id)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@router.patch("/{user_id}", summary="Update user", response_model=UserResponse)
+async def update_user(
+    user_id: int,
+    payload: UserUpdate,
+    user_service=Depends(get_user_service),
+):
+    try:
+        return await user_service.update_user(user_id, payload)
+    except Exception as exc:  # pragma: no cover
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+

backend/api/mock_api_route.py

+import asyncio
+import json
+import logging
+import time
+
+from fastapi import APIRouter, BackgroundTasks, HTTPException
+from pydantic import BaseModel
+
+from agent.tools.data_retrieval_tool import SearchItem, data_retrieval_tool
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+# --- HELPERS ---
+async def retry_with_backoff(coro_fn, max_retries=3, backoff_factor=2):
+    """Retry async function with exponential backoff"""
+    for attempt in range(max_retries):
+        try:
+            return await coro_fn()
+        except Exception as e:
+            if attempt == max_retries - 1:
+                raise
+            wait_time = backoff_factor**attempt
+            logger.warning(f"⚠️ Attempt {attempt + 1} failed: {e!s}, retrying in {wait_time}s...")
+            await asyncio.sleep(wait_time)
+
+
+# --- MODELS ---
+class MockQueryRequest(BaseModel):
+    user_query: str
+    user_id: str | None = "test_user"
+    session_id: str | None = None
+
+
+class MockDBRequest(BaseModel):
+    query: str | None = None
+    magento_ref_code: str | None = None
+    price_min: float | None = None
+    price_max: float | None = None
+    top_k: int = 10
+
+
+class MockRetrieverRequest(BaseModel):
+    user_query: str
+    price_min: float | None = None
+    price_max: float | None = None
+    magento_ref_code: str | None = None
+    user_id: str | None = "test_user"
+    session_id: str | None = None
+
+
+# --- MOCK LLM RESPONSES (không gọi OpenAI) ---
+MOCK_AI_RESPONSES = [
+    "Dựa trên tìm kiếm của bạn, tôi tìm thấy các sản phẩm phù hợp với nhu cầu của bạn. Những mặt hàng này có chất lượng tốt và giá cả phải chăng.",
+    "Tôi gợi ý cho bạn những sản phẩm sau. Chúng đều là những lựa chọn phổ biến và nhận được đánh giá cao từ khách hàng.",
+    "Dựa trên tiêu chí tìm kiếm của bạn, đây là những sản phẩm tốt nhất mà tôi có thể giới thiệu.",
+    "Những sản phẩm này hoàn toàn phù hợp với yêu cầu của bạn. Hãy xem chi tiết để chọn sản phẩm yêu thích nhất.",
+    "Tôi đã tìm được các mặt hàng tuyệt vời cho bạn. Hãy kiểm tra chúng để tìm ra lựa chọn tốt nhất.",
+]
+
+
+# --- ENDPOINTS ---
+
+
+from agent.mock_controller import mock_chat_controller
+
+
+@router.post("/mock/agent/chat", summary="Mock Agent Chat (Real Tools + Fake LLM)")
+async def mock_chat(req: MockQueryRequest, background_tasks: BackgroundTasks):
+    """
+    Mock Agent Chat using mock_chat_controller:
+    - ✅ Real embedding + vector search (data_retrieval_tool THẬT)
+    - ✅ Real products from StarRocks
+    - ❌ Fake LLM response (no OpenAI cost)
+    - Perfect for stress testing + end-to-end testing
+    """
+    try:
+        logger.info(f"🚀 [Mock Agent Chat] Starting with query: {req.user_query}")
+
+        result = await mock_chat_controller(
+            query=req.user_query,
+            user_id=req.user_id or "test_user",
+            background_tasks=background_tasks,
+        )
+
+        return {
+            "status": "success",
+            "user_query": req.user_query,
+            "user_id": req.user_id,
+            "session_id": req.session_id,
+            **result,  # Include status, ai_response, product_ids, etc.
+        }
+
+    except Exception as e:
+        logger.error(f"❌ Error in mock agent chat: {e!s}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Mock Agent Chat Error: {e!s}")
+
+
+
+
+@router.post("/mock/db/search", summary="Real Data Retrieval Tool (Agent Tool)")
+async def mock_db_search(req: MockDBRequest):
+    """
+    Dùng `data_retrieval_tool` THẬT từ Agent:
+    - Nếu có magento_ref_code → CODE SEARCH (không cần embedding)
+    - Nếu có query → HYDE SEMANTIC SEARCH (embedding + vector search)
+    - Lọc theo giá nếu có price_min/price_max
+    - Trả về sản phẩm thực từ StarRocks
+
+    Format input giống SearchItem của agent tool.
+    """
+    try:
+        logger.info("📍 Data Retrieval Tool called")
+        start_time = time.time()
+
+        # Xây dựng SearchItem từ request
+        search_item = SearchItem(
+            query=req.query or "sản phẩm",
+            magento_ref_code=req.magento_ref_code,
+            price_min=req.price_min,
+            price_max=req.price_max,
+            action="search",
+        )
+
+        logger.info(f"🔧 Search params: {search_item.dict(exclude_none=True)}")
+
+        # Gọi data_retrieval_tool THẬT với retry
+        result_json = await retry_with_backoff(
+            lambda: data_retrieval_tool.ainvoke({"searches": [search_item]}), max_retries=3
+        )
+        result = json.loads(result_json)
+
+        elapsed_time = time.time() - start_time
+        logger.info(f"✅ Data Retrieval completed in {elapsed_time:.3f}s")
+
+        return {
+            "status": result.get("status", "success"),
+            "search_params": search_item.dict(exclude_none=True),
+            "total_results": len(result.get("results", [{}])[0].get("products", [])),
+            "products": result.get("results", [{}])[0].get("products", []),
+            "processing_time_ms": round(elapsed_time * 1000, 2),
+            "raw_result": result,
+        }
+
+    except Exception as e:
+        logger.error(f"❌ Error in DB search: {e!s}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"DB Search Error: {e!s}")
+
+
+@router.post("/mock/retriverdb", summary="Real Embedding + Real DB Vector Search")
+async def mock_retriever_db(req: MockRetrieverRequest):
+    """
+    API thực tế để test Retriever + DB Search (dùng agent tool):
+    - Lấy query từ user
+    - Embedding THẬT (gọi OpenAI embedding trong tool)
+    - Vector search THẬT trong StarRocks
+    - Trả về kết quả sản phẩm thực (bỏ qua LLM)
+
+    Dùng để test performance của embedding + vector search riêng biệt.
+    """
+    try:
+        logger.info(f"📍 Retriever DB started: {req.user_query}")
+        start_time = time.time()
+
+        # Xây dựng SearchItem từ request
+        search_item = SearchItem(
+            query=req.user_query,
+            magento_ref_code=req.magento_ref_code,
+            price_min=req.price_min,
+            price_max=req.price_max,
+            action="search",
+        )
+
+        logger.info(f"🔧 Retriever params: {search_item.dict(exclude_none=True)}")
+
+        # Gọi data_retrieval_tool THẬT (embedding + vector search) với retry
+        result_json = await retry_with_backoff(
+            lambda: data_retrieval_tool.ainvoke({"searches": [search_item]}), max_retries=3
+        )
+        result = json.loads(result_json)
+
+        elapsed_time = time.time() - start_time
+        logger.info(f"✅ Retriever completed in {elapsed_time:.3f}s")
+
+        # Parse kết quả
+        search_results = result.get("results", [{}])[0]
+        products = search_results.get("products", [])
+
+        return {
+            "status": result.get("status", "success"),
+            "user_query": req.user_query,
+            "user_id": req.user_id,
+            "session_id": req.session_id,
+            "search_params": search_item.dict(exclude_none=True),
+            "total_results": len(products),
+            "products": products,
+            "processing_time_ms": round(elapsed_time * 1000, 2),
+        }
+
+    except Exception as e:
+        logger.error(f"❌ Error in retriever DB: {e!s}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Retriever DB Error: {e!s}")

backend/api/prompt_route.py

+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+import os
+from agent.graph import reset_graph
+
+router = APIRouter()
+
+PROMPT_FILE_PATH = os.path.join(os.path.dirname(__file__), "../agent/system_prompt.txt")
+
+class PromptUpdateRequest(BaseModel):
+    content: str
+
+@router.get("/api/agent/system-prompt")
+async def get_system_prompt_content():
+    """Get current system prompt content"""
+    # ... existing code ...
+    try:
+        if os.path.exists(PROMPT_FILE_PATH):
+            with open(PROMPT_FILE_PATH, "r", encoding="utf-8") as f:
+                content = f.read()
+            return {"status": "success", "content": content}
+        else:
+            return {"status": "error", "message": "Prompt file not found"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/api/agent/system-prompt")
+async def update_system_prompt_content(request: PromptUpdateRequest):
+    """Update system prompt content"""
+    try:
+        # 1. Update file
+        with open(PROMPT_FILE_PATH, "w", encoding="utf-8") as f:
+            f.write(request.content)
+            
+        # 2. Reset Graph Singleton to force reload prompt
+        reset_graph()
+        
+        return {"status": "success", "message": "System prompt updated successfully. Graph reloaded."}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

backend/common/__init__.py

+

backend/common/cache.py

+import hashlib
+import json
+import logging
+
+import redis.asyncio as aioredis  # redis package với async support (thay thế aioredis deprecated)
+
+from config import (
+    REDIS_CACHE_DB,
+    REDIS_CACHE_PORT,
+    REDIS_CACHE_TURN_ON,
+    REDIS_CACHE_URL,
+    REDIS_PASSWORD,
+    REDIS_USERNAME,
+)
+
+logger = logging.getLogger(__name__)
+
+# ====================== CACHE CONFIGURATION ======================
+# Layer 1: Response Cache (Short TTL to keep stock/price safe)
+DEFAULT_RESPONSE_TTL = 300  # 5 minutes
+RESPONSE_KEY_PREFIX = "resp_cache:"
+
+# Layer 2: Embedding Cache (Long TTL since vectors are static)
+EMBEDDING_CACHE_TTL = 86400  # 24 hours
+EMBEDDING_KEY_PREFIX = "emb_cache:"
+
+class RedisClient:
+    """
+    Hybrid Cache Client for Canifa Chatbot.
+    Layer 1: Exact Response Cache (Short TTL)
+    Layer 2: Embedding Cache (Long TTL)
+    """
+
+    def __init__(self):
+        self._client: aioredis.Redis | None = None
+        self._enabled = REDIS_CACHE_TURN_ON
+        self._stats = {
+            "resp_hits": 0,
+            "emb_hits": 0,
+            "misses": 0,
+        }
+
+    async def initialize(self) -> aioredis.Redis | None:
+        """Initialize connection"""
+        if not self._enabled:
+            logger.info("🚫 Redis Cache is DISABLED via REDIS_CACHE_TURN_ON")
+            return None
+
+        if self._client is not None:
+            return self._client
+
+        try:
+            connection_kwargs = {
+                "host": REDIS_CACHE_URL,
+                "port": REDIS_CACHE_PORT,
+                "db": REDIS_CACHE_DB,
+                "decode_responses": True,
+                "socket_connect_timeout": 5,
+            }
+            if REDIS_PASSWORD:
+                connection_kwargs["password"] = REDIS_PASSWORD
+            if REDIS_USERNAME:
+                connection_kwargs["username"] = REDIS_USERNAME
+
+            self._client = aioredis.Redis(**connection_kwargs)
+            await self._client.ping()
+
+            logger.info(f"✅ Redis Hybrid Cache connected: {REDIS_CACHE_URL}:{REDIS_CACHE_PORT} (db={REDIS_CACHE_DB})")
+            return self._client
+
+        except Exception as e:
+            logger.error(f"❌ Failed to connect to Redis: {e}")
+            self._enabled = False
+            return None
+
+    def get_client(self) -> aioredis.Redis | None:
+        if not self._enabled:
+            return None
+        return self._client
+
+    # --- Layer 1: Exact Response Cache (Short TTL) ---
+
+    async def get_response(self, user_id: str, query: str) -> dict | None:
+        """Get exact matched response (100% safe, short TTL)"""
+        if not self._enabled: return None
+        try:
+            client = self.get_client()
+            if not client: return None
+
+            # Hash of (user_id + query) for exact match
+            query_key = f"{user_id}:{query.strip().lower()}"
+            cache_hash = hashlib.md5(query_key.encode()).hexdigest()
+            key = f"{RESPONSE_KEY_PREFIX}{cache_hash}"
+
+            cached = await client.get(key)
+            if cached:
+                self._stats["resp_hits"] += 1
+                logger.info(f"⚡ LAYER 1 HIT (Response) | User: {user_id}")
+                return json.loads(cached)
+
+            return None
+        except Exception as e:
+            logger.warning(f"Redis get_response error: {e}")
+            return None
+
+    async def set_response(self, user_id: str, query: str, response_data: dict, ttl: int = DEFAULT_RESPONSE_TTL):
+        """Store full response in cache with short TTL"""
+        if not self._enabled or not response_data: return
+        try:
+            client = self.get_client()
+            if not client: return
+
+            query_key = f"{user_id}:{query.strip().lower()}"
+            cache_hash = hashlib.md5(query_key.encode()).hexdigest()
+            key = f"{RESPONSE_KEY_PREFIX}{cache_hash}"
+
+            await client.setex(key, ttl, json.dumps(response_data))
+            logger.debug(f"💾 LAYER 1 STORED (Response) | TTL: {ttl}s")
+        except Exception as e:
+            logger.warning(f"Redis set_response error: {e}")
+
+    # --- Layer 2: Embedding Cache (Long TTL) ---
+
+    async def get_embedding(self, text: str) -> list[float] | None:
+        """Get cached embedding (Saves OpenAI costs)"""
+        if not self._enabled: return None
+        try:
+            client = self.get_client()
+            if not client: return None
+
+            text_hash = hashlib.md5(text.strip().lower().encode()).hexdigest()
+            key = f"{EMBEDDING_KEY_PREFIX}{text_hash}"
+
+            cached = await client.get(key)
+            if cached:
+                self._stats["emb_hits"] += 1
+                logger.info(f"🔵 LAYER 2 HIT (Embedding) | Query: {text[:20]}...")
+                return json.loads(cached)
+
+            return None
+        except Exception as e:
+            logger.warning(f"Redis get_embedding error: {e}")
+            return None
+
+    async def set_embedding(self, text: str, embedding: list[float], ttl: int = EMBEDDING_CACHE_TTL):
+        """Store embedding for long term"""
+        if not self._enabled or not embedding: return
+        try:
+            client = self.get_client()
+            if not client: return
+
+            text_hash = hashlib.md5(text.strip().lower().encode()).hexdigest()
+            key = f"{EMBEDDING_KEY_PREFIX}{text_hash}"
+
+            await client.setex(key, ttl, json.dumps(embedding))
+            logger.debug(f"💾 LAYER 2 STORED (Embedding) | TTL: {ttl}s")
+        except Exception as e:
+            logger.warning(f"Redis set_embedding error: {e}")
+
+# --- Singleton Export ---
+redis_cache = RedisClient()
+
+def  get_redis_cache() -> RedisClient:
+    return redis_cache

backend/common/canifa_api.py

+"""
+Canifa API Service
+Xử lý các logic liên quan đến API của Canifa (Magento)
+"""
+import logging
+import httpx
+from typing import Optional, Dict, Any
+
+logger = logging.getLogger(__name__)
+
+# URL API Canifa
+CANIFA_CUSTOMER_API = "https://canifa.com/v1/magento/customer"
+
+# GraphQL Query Body giả lập (để lấy User Info)
+CANIFA_QUERY_BODY = [
+    {
+        "customer": "customer-custom-query",
+        "metadata": {
+            "fields": "\n customer {\n gender\n customer_id\n phone_number\n date_of_birth\n default_billing\n default_shipping\n email\n firstname\n is_subscribed\n lastname\n middlename\n prefix\n suffix\n taxvat\n addresses {\n city\n country_code\n default_billing\n default_shipping\n extension_attributes {\n attribute_code\n value\n }\n custom_attributes {\n attribute_code\n value\n }\n firstname\n id\n lastname\n postcode\n prefix\n region {\n region_code\n region_id\n region\n }\n street\n suffix\n telephone\n vat_id\n }\n is_subscribed\n }\n "
+        }
+    },
+    {}
+]
+
+async def verify_canifa_token(token: str) -> Optional[Dict[str, Any]]:
+    """
+    Verify token với API Canifa (Magento).
+    Dùng token làm cookie `vsf-customer` để gọi API lấy thông tin customer.
+    
+    Args:
+        token: Giá trị của cookie vsf-customer (lấy từ Header Authorization)
+        
+    Returns:
+        Dict info user hoặc None nếu lỗi
+    """
+    if not token:
+        return None
+        
+    headers = {
+        "accept": "application/json, text/plain, */*",
+        "content-type": "application/json",
+        "Cookie": f"vsf-customer={token}"  # Quan trọng: Gửi token dưới dạng Cookie
+    }
+    
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.post(
+                CANIFA_CUSTOMER_API,
+                json=CANIFA_QUERY_BODY,
+                headers=headers
+            )
+            
+            if response.status_code == 200:
+                data = response.json()
+                logger.debug(f"Canifa API Raw Response: {data}")
+                
+                # Response format: {"data": {"customer": {...}}, "loading": false, ...}
+                if isinstance(data, dict):
+                    # Trả về toàn bộ data để extract_user_id xử lý
+                    return data
+                
+                # Nếu Canifa trả list (batch request)
+                return data
+
+            else:
+                logger.warning(f"Canifa API Failed: {response.status_code} - {response.text}")
+                return None
+                
+    except Exception as e:
+        logger.error(f"Error calling Canifa API: {e}")
+        return None
+
+async def extract_user_id_from_canifa_response(data: Any) -> Optional[str]:
+    """
+    Bóc customer_id từ response data của Canifa.
+    """
+    if not data:
+        return None
+        
+    try:
+        # Dự phòng các format data trả về khác nhau
+        customer = None
+        
+        # Format 1: data['customer']
+        if isinstance(data, dict):
+            customer = data.get('customer') or data.get('data', {}).get('customer')
+            
+        # Format 2: data là list (nếu query batch)
+        elif isinstance(data, list) and len(data) > 0:
+            item = data[0]
+            if isinstance(item, dict):
+                customer = item.get('result', {}).get('customer') or item.get('data', {}).get('customer')
+
+        if customer and isinstance(customer, dict):
+            user_id = customer.get('customer_id') or customer.get('id')
+            if user_id:
+                return str(user_id)
+                
+        return None
+        
+    except Exception as e:
+        logger.error(f"Error parsing user_id from Canifa response: {e}")
+        return None

backend/common/datadb/add_context_to_sizes.py

+"""
+Script tự động thêm context (tên bảng + subsection) vào tất cả size entries trong tonghop.txt
+Ví dụ: "Size 92 (2Y):" -> "Size 92 (2Y) - BẢNG SIZE CHUNG CHO UNISEX - TRẺ EM (Dải size lẻ):"
+"""
+
+import re
+
+
+def add_context_to_sizes(input_file, output_file):
+    with open(input_file, encoding="utf-8") as f:
+        lines = f.readlines()
+
+    result = []
+    current_table = None  # Tên bảng hiện tại
+    current_subsection = None  # Subsection hiện tại (Dải size lẻ, chẵn...)
+
+    for line in lines:
+        stripped = line.strip()
+
+        # Phát hiện header bảng (bắt đầu bằng BẢNG hoặc QUẦN)
+        if stripped.startswith("BẢNG SIZE") or stripped.startswith("QUẦN"):
+            current_table = stripped
+            current_subsection = None  # Reset subsection khi sang bảng mới
+            result.append(line)
+            continue
+
+        # Phát hiện subsection (Dải size lẻ, Dải size chẵn)
+        if "Dải size" in stripped or stripped.startswith("Dải"):
+            current_subsection = stripped.rstrip(":")
+            result.append(line)
+            continue
+
+        # Phát hiện dòng Size (bắt mọi pattern: Size XS:, Size 92 (2Y):, Size 26 (XS):)
+        size_match = re.match(r"^(Size\s+[A-Z0-9]+(?:\s*\([^)]+\))?):(.*)$", stripped)
+        if size_match and current_table:
+            size_part = size_match.group(1)  # "Size 92 (2Y)" hoặc "Size XS"
+            rest = size_match.group(2)  # Phần còn lại sau dấu :
+
+            # Xây dựng context
+            context_parts = [current_table]
+            if current_subsection:
+                context_parts.append(f"({current_subsection})")
+
+            context = " - ".join(context_parts)
+
+            # Tạo dòng mới với context
+            new_line = f"{size_part} - {context}:{rest}\n"
+            result.append(new_line)
+            continue
+
+        # Giữ nguyên các dòng khác
+        result.append(line)
+
+    # Ghi file output
+    with open(output_file, "w", encoding="utf-8") as f:
+        f.writelines(result)
+
+    print("✅ Đã thêm context vào tất cả size entries!")
+    print(f"📝 File output: {output_file}")
+
+
+if __name__ == "__main__":
+    input_path = r"d:\cnf\chatbot_canifa\backend\datadb\tonghop.txt"
+    output_path = r"d:\cnf\chatbot_canifa\backend\datadb\tonghop_with_context.txt"
+
+    add_context_to_sizes(input_path, output_path)
+
+    print("\n🔍 Preview 10 dòng đầu của file mới:")
+    with open(output_path, encoding="utf-8") as f:
+        for i, line in enumerate(f):
+            if i >= 1160 and i < 1170:  # Vùng có size entries
+                print(line.rstrip())

backend/common/datadb/ingest_knowledge.py

+import os
+import json
+import pymysql
+from openai import OpenAI
+import time
+
+# ==========================================
+# 🔐 HARD KEY CONFIGURATION (As requested)
+# ==========================================
+OPENAI_API_KEY = "sk-proj-srJ3l3B5q1CzRezXAnaewbbRfuWzIjYHbcAdggzsa4MmtXEHaIwS1OTkMgLpMDikgh"
+SR_HOST = "172.16.2.100"
+SR_PORT = 9030
+SR_USER = "anhvh"
+SR_PASS = "v0WYGeyLRCckXotT"
+SR_DB = "shared_source"
+
+# Parameter
+CHUNK_SIZE = 500
+CHUNK_OVERLAP = 50
+EMBEDDING_MODEL = "text-embedding-3-small" # 1536 dimensions
+
+client = OpenAI(api_key=OPENAI_API_KEY)
+
+def get_embedding(text):
+    """Lấy vector 1536 chiều từ OpenAI"""
+    try:
+        text = text.replace("\n", " ")
+        return client.embeddings.create(input=[text], model=EMBEDDING_MODEL).data[0].embedding
+    except Exception as e:
+        print(f"❌ Lỗi Embedding: {e}")
+        return None
+
+def connect_starrocks():
+    return pymysql.connect(
+        host=SR_HOST,
+        port=SR_PORT,
+        user=SR_USER,
+        password=SR_PASS,
+        database=SR_DB,
+        charset='utf8mb4',
+        cursorclass=pymysql.cursors.DictCursor
+    )
+
+def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
+    """Chia nhỏ văn bản với overlap"""
+    chunks = []
+    start = 0
+    while start < len(text):
+        end = start + size
+        chunks.append(text[start:end])
+        start += size - overlap
+    return chunks
+
+def ingest():
+    input_file = r"d:\cnf\chatbot_canifa\backend\datadb\tonghop.txt"
+    if not os.path.exists(input_file):
+        print(f"❌ Không tìm thấy file: {input_file}")
+        return
+
+    print(f"📖 Đang đọc file {input_file}...")
+    with open(input_file, "r", encoding="utf-8") as f:
+        full_content = f.read()
+
+    # Tách dữ liệu theo từng FILE giả định trong tonghop.txt
+    sections = full_content.split("================================================================================")
+    
+    db = connect_starrocks()
+    cursor = db.cursor()
+    
+    total_chunks = 0
+    record_id = int(time.time()) # Làm ID cơ bản
+
+    for section in sections:
+        if not section.strip(): continue
+        
+        # Lấy tiêu đề file nếu có
+        lines = section.strip().split("\n")
+        title = "Canifa Knowledge"
+        if "FILE:" in lines[0]:
+            title = lines[0].replace("FILE:", "").strip()
+            content = "\n".join(lines[1:])
+        else:
+            content = section
+
+        print(f"🚀 Đang xử lý section: {title}")
+        chunks = chunk_text(content)
+        
+        for i, chunk in enumerate(chunks):
+            if len(chunk.strip()) < 20: continue # Bỏ qua đoạn quá ngắn
+            
+            vector = get_embedding(chunk)
+            if not vector: continue
+            
+            metadata = {
+                "title": title,
+                "chunk_idx": i,
+                "source": "tonghop.txt",
+                "timestamp": time.time()
+            }
+            
+            sql = "INSERT INTO shared_source.canifa_knowledge (id, content, metadata, embedding) VALUES (%s, %s, %s, %s)"
+            try:
+                cursor.execute(sql, (record_id, chunk, json.dumps(metadata, ensure_ascii=False), str(vector)))
+                record_id += 1
+                total_chunks += 1
+                if total_chunks % 10 == 0:
+                    db.commit()
+                    print(f"✅ Đã nạp {total_chunks} chunks...")
+            except Exception as e:
+                print(f"❌ Lỗi SQL: {e}")
+                
+    db.commit()
+    db.close()
+    print(f"🎊 HOÀN THÀNH! Tổng cộng đã nạp {total_chunks} vào StarRocks.")
+
+if __name__ == "__main__":
+    ingest()

backend/common/embedding_service.py

+import logging
+
+from openai import AsyncOpenAI, OpenAI
+
+from config import OPENAI_API_KEY
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "create_embedding",
+    "create_embedding_async",
+    "create_embeddings_async",
+    "get_async_embedding_client",
+    "get_embedding_client",
+]
+
+
+class EmbeddingClientManager:
+    """
+    Singleton Class quản lý OpenAI Embedding Client (Sync & Async).
+    """
+
+    def __init__(self):
+        self._client: OpenAI | None = None
+        self._async_client: AsyncOpenAI | None = None
+
+    def get_client(self) -> OpenAI:
+        """Sync Client lazy loading"""
+        if self._client is None:
+            if not OPENAI_API_KEY:
+                raise RuntimeError("CRITICAL: OPENAI_API_KEY chưa được thiết lập")
+            self._client = OpenAI(api_key=OPENAI_API_KEY)
+        return self._client
+
+    def get_async_client(self) -> AsyncOpenAI:
+        """Async Client lazy loading"""
+        if self._async_client is None:
+            if not OPENAI_API_KEY:
+                raise RuntimeError("CRITICAL: OPENAI_API_KEY chưa được thiết lập")
+            self._async_client = AsyncOpenAI(api_key=OPENAI_API_KEY)
+        return self._async_client
+
+
+logger = logging.getLogger(__name__)
+
+# NOTE:
+# - TẠM THỜI KHÔNG DÙNG REDIS CACHE CHO EMBEDDING để tránh phụ thuộc Redis/aioredis.
+# - Nếu cần bật lại cache, import `redis_cache` từ `common.cache`
+#   và dùng như các đoạn code cũ (get_embedding / set_embedding).
+
+# --- Singleton ---
+_manager = EmbeddingClientManager()
+get_embedding_client = _manager.get_client
+get_async_embedding_client = _manager.get_async_client
+
+
+def create_embedding(text: str) -> list[float]:
+    """Sync embedding generation (No cache for sync to avoid overhead)"""
+    try:
+        client = get_embedding_client()
+        response = client.embeddings.create(model="text-embedding-3-small", input=text)
+        return response.data[0].embedding
+    except Exception as e:
+        logger.error(f"Error creating embedding (sync): {e}")
+        return []
+
+
+async def create_embedding_async(text: str) -> list[float]:
+    """
+    Async embedding generation (KHÔNG dùng cache).
+    Nếu sau này cần cache lại, có thể thêm redis_cache.get_embedding / set_embedding.
+    """
+    try:
+        client = get_async_embedding_client()
+        response = await client.embeddings.create(model="text-embedding-3-small", input=text)
+        embedding = response.data[0].embedding
+        return embedding
+    except Exception as e:
+        logger.error(f"Error creating embedding (async): {e}")
+        return []
+
+
+async def create_embeddings_async(texts: list[str]) -> list[list[float]]:
+    """
+    Batch async embedding generation with per-item Layer 2 Cache.
+    """
+    try:
+        if not texts:
+            return []
+
+        client = get_async_embedding_client()
+        response = await client.embeddings.create(model="text-embedding-3-small", input=texts)
+
+        # Giữ nguyên thứ tự embedding theo order input
+        sorted_data = sorted(response.data, key=lambda x: x.index)
+        results = [item.embedding for item in sorted_data]
+
+        return results
+    except Exception as e:
+        logger.error(f"Error creating batch embeddings (async): {e}")
+        return [[] for _ in texts]

backend/common/image_storage.py

+import logging
+import uuid
+
+import httpx
+
+from config import CONV_SUPABASE_KEY, CONV_SUPABASE_URL
+
+logger = logging.getLogger(__name__)
+
+
+class ImageStorageService:
+    """
+    Service quản lý việc upload ảnh lên Supabase Storage.
+    Sử dụng httpx để gọi REST API của Supabase.
+    """
+
+    def __init__(self, bucket_name: str = "chat-images"):
+        self.url = CONV_SUPABASE_URL
+        self.key = CONV_SUPABASE_KEY
+        self.bucket_name = bucket_name
+
+        if not self.url or not self.key:
+            logger.error("Supabase URL or Key is missing in configuration")
+
+    async def upload_base64(self, base64_data: str, filename: str | None = None) -> str | None:
+        """
+        Upload ảnh từ base64 string lên Supabase Storage.
+        Returns: Public URL của ảnh hoặc None nếu lỗi.
+        """
+        try:
+            if not self.url or not self.key:
+                return None
+
+            # Xử lý base64 string (loại bỏ prefix nếu có)
+            if "," in base64_data:
+                header, base64_data = base64_data.split(",", 1)
+                content_type = header.split(";")[0].split(":")[1]
+            else:
+                content_type = "image/jpeg"
+
+            import base64
+
+            file_content = base64.b64decode(base64_data)
+
+            if not filename:
+                ext = content_type.split("/")[-1]
+                filename = f"{uuid.uuid4()}.{ext}"
+
+            # Supabase Storage REST API: /storage/v1/object/{bucket}/{path}
+            upload_url = f"{self.url}/storage/v1/object/{self.bucket_name}/{filename}"
+
+            headers = {"Authorization": f"Bearer {self.key}", "apikey": self.key, "Content-Type": content_type}
+
+            async with httpx.AsyncClient() as client:
+                response = await client.post(upload_url, content=file_content, headers=headers)
+
+                if response.status_code == 200:
+                    # Lấy public URL (Giả định bucket là public)
+                    public_url = f"{self.url}/storage/v1/object/public/{self.bucket_name}/{filename}"
+                    logger.info(f"✅ Uploaded image successfully: {public_url}")
+                    return public_url
+                logger.error(f"❌ Failed to upload image: {response.status_code} - {response.text}")
+                return None
+
+        except Exception as e:
+            logger.error(f"Error uploading image to Supabase: {e}")
+            return None

backend/common/langfuse_client.py

+"""
+Simple Langfuse Client Wrapper
+Minimal setup using langfuse.langchain module
+With propagate_attributes for proper user_id tracking
+"""
+
+import asyncio
+import logging
+from concurrent.futures import ThreadPoolExecutor
+
+from langfuse import Langfuse, get_client
+from langfuse.langchain import CallbackHandler
+
+from config import (
+    LANGFUSE_PUBLIC_KEY,
+    LANGFUSE_SECRET_KEY,
+)
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["async_flush_langfuse", "get_callback_handler", "get_langfuse_client"]
+
+
+class LangfuseClientManager:
+    """
+    Singleton manager for Langfuse client.
+    Lazy loading - only initialize when first needed.
+    """
+
+    def __init__(self):
+        self._client: Langfuse | None = None
+        self._executor: ThreadPoolExecutor | None = None
+        self._initialized = False
+
+    def get_client(self) -> Langfuse | None:
+        """
+        Lazy loading - initialize Langfuse client on first call.
+        """
+        if self._initialized:
+            return self._client
+
+        logger.info("🔧 [LAZY LOADING] Initializing Langfuse client (first time)")
+
+        if not LANGFUSE_PUBLIC_KEY or not LANGFUSE_SECRET_KEY:
+            logger.warning("⚠️ LANGFUSE KEYS MISSING. Tracing disabled.")
+            self._initialized = True
+            return None
+
+        try:
+            self._client = get_client()
+            self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="langfuse_export")
+
+            if self._client.auth_check():
+                logger.info("✅ Langfuse Ready! (async batch export)")
+                self._initialized = True
+                return self._client
+
+            logger.error("❌ Langfuse auth failed")
+            self._initialized = True
+            return None
+
+        except Exception as e:
+            logger.error(f"❌ Langfuse init error: {e}")
+            self._initialized = True
+            return None
+
+    async def async_flush(self):
+        """
+        Async wrapper to flush Langfuse without blocking event loop.
+        Uses thread pool executor to run sync flush in background.
+        """
+        client = self.get_client()
+        if not client or not self._executor:
+            return
+
+        try:
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(self._executor, client.flush)
+            logger.debug("📤 Langfuse flushed (async)")
+        except Exception as e:
+            logger.warning(f"⚠️ Async flush failed: {e}")
+
+    def get_callback_handler(self) -> CallbackHandler | None:
+        """Get CallbackHandler instance."""
+        client = self.get_client()
+        if not client:
+            logger.warning("⚠️ Langfuse client not available")
+            return None
+
+        try:
+            handler = CallbackHandler()
+            logger.debug("✅ Langfuse CallbackHandler created")
+            return handler
+        except Exception as e:
+            logger.warning(f"⚠️ CallbackHandler error: {e}")
+            return None
+
+
+# --- Singleton ---
+_manager = LangfuseClientManager()
+get_langfuse_client = _manager.get_client
+async_flush_langfuse = _manager.async_flush
+
+
+def get_callback_handler() -> CallbackHandler | None:
+    """Get CallbackHandler instance (wrapper for manager)."""
+    return _manager.get_callback_handler()

backend/common/llm_factory.py

+"""
+LLM Factory - OpenAI LLM creation with caching.
+Manages initialization and caching of OpenAI models.
+"""
+
+import contextlib
+import logging
+
+from langchain_core.language_models import BaseChatModel
+from langchain_openai import ChatOpenAI, OpenAIEmbeddings
+
+from config import OPENAI_API_KEY
+
+logger = logging.getLogger(__name__)
+
+
+class LLMFactory:
+    """Singleton factory for managing OpenAI LLM instances with caching."""
+
+    COMMON_MODELS: list[str] = [
+        "gpt-4o-mini",
+        "gpt-4o",
+        "gpt-5-nano",
+        "gpt-5-mini",
+    ]
+
+    def __init__(self):
+        """Initialize LLM factory with empty cache."""
+        self._cache: dict[tuple[str, bool, bool, str | None], BaseChatModel] = {}
+
+    def get_model(
+        self,
+        model_name: str,
+        streaming: bool = True,
+        json_mode: bool = False,
+        api_key: str | None = None,
+    ) -> BaseChatModel:
+        """
+        Get or create an LLM instance from cache.
+
+        Args:
+            model_name: Model identifier (e.g., "gpt-4o-mini", "gemini-2.0-flash-lite-preview-02-05")
+            streaming: Enable streaming responses
+            json_mode: Enable JSON output format
+            api_key: Optional API key override
+
+        Returns:
+            Configured LLM instance
+        """
+        clean_model = model_name.split("/")[-1] if "/" in model_name else model_name
+        cache_key = (clean_model, streaming, json_mode, api_key)
+
+        if cache_key in self._cache:
+            logger.debug(f"♻️ Using cached model: {clean_model}")
+            return self._cache[cache_key]
+
+        logger.info(f"Creating new LLM instance: {clean_model}")
+        return self._create_instance(clean_model, streaming, json_mode, api_key)
+
+    def _create_instance(
+        self,
+        model_name: str,
+        streaming: bool = False,
+        json_mode: bool = False,
+        api_key: str | None = None,
+    ) -> BaseChatModel:
+        """Create and cache a new OpenAI LLM instance."""
+        try:
+            llm = self._create_openai(model_name, streaming, json_mode, api_key)
+            
+            cache_key = (model_name, streaming, json_mode, api_key)
+            self._cache[cache_key] = llm
+            return llm
+
+        except Exception as e:
+            logger.error(f"❌ Failed to create model {model_name}: {e}")
+            raise
+
+    def _create_openai(self, model_name: str, streaming: bool, json_mode: bool, api_key: str | None) -> BaseChatModel:
+        """Create OpenAI model instance."""
+        key = api_key or OPENAI_API_KEY
+
+        if not key:
+            raise ValueError("OPENAI_API_KEY is required")
+
+        llm_kwargs = {
+            "model": model_name,
+            "streaming": streaming,
+            "api_key": key,
+            "temperature": 0,
+            "max_tokens": 1000,
+        }
+        
+        # Nếu bật json_mode, tiêm trực tiếp vào constructor
+        if json_mode:
+            llm_kwargs["model_kwargs"] = {"response_format": {"type": "json_object"}}
+            logger.info(f"⚙️ Initializing OpenAI in JSON mode: {model_name}")
+
+        llm = ChatOpenAI(**llm_kwargs)
+        logger.info(f"✅ Created OpenAI: {model_name}")
+        return llm
+
+    def _enable_json_mode(self, llm: BaseChatModel, model_name: str) -> BaseChatModel:
+        """Enable JSON mode for the LLM."""
+        try:
+            llm = llm.bind(response_format={"type": "json_object"})
+            logger.debug(f"⚙️ JSON mode enabled for {model_name}")
+        except Exception as e:
+            logger.warning(f"⚠️ JSON mode not supported: {e}")
+        return llm
+
+    def initialize(self, skip_warmup: bool = True) -> None:
+        """
+        Pre-initialize common models.
+
+        Args:
+            skip_warmup: Skip initialization if True
+        """
+        if skip_warmup or self._cache:
+            return
+
+        logger.info("🔥 Warming up LLM Factory...")
+        for model_name in self.COMMON_MODELS:
+            with contextlib.suppress(Exception):
+                self.get_model(model_name, streaming=True)
+
+
+# --- Singleton Instance & Public API ---
+_factory = LLMFactory()
+
+
+def create_llm(
+    model_name: str,
+    streaming: bool = True,
+    json_mode: bool = False,
+    api_key: str | None = None,
+) -> BaseChatModel:
+    """Create or get cached LLM instance."""
+    return _factory.get_model(model_name, streaming=streaming, json_mode=json_mode, api_key=api_key)
+
+
+def init_llm_factory(skip_warmup: bool = True) -> None:
+    """Initialize the LLM factory."""
+    _factory.initialize(skip_warmup)
+
+
+def create_embedding_model() -> OpenAIEmbeddings:
+    """Create OpenAI embeddings model."""
+    return OpenAIEmbeddings(model="text-embedding-3-small", api_key=OPENAI_API_KEY)

backend/common/openai_client.py

+# app/common/openai_client.py
+import logging
+
+from openai import AsyncOpenAI
+
+from config import OPENAI_API_KEY
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["get_openai_client", "init_openai_client"]
+
+
+class OpenAIClientManager:
+    """
+    Singleton Class quản lý OpenAI Client.
+    Đảm bảo chỉ có 1 instance AsyncOpenAI được tạo ra trong toàn bộ ứng dụng.
+    """
+
+    def __init__(self):
+        self._client: AsyncOpenAI | None = None
+
+    def initialize(self) -> None:
+        """
+        Khởi tạo AsyncOpenAI client.
+        Gọi 1 lần ở lifespan/startup.
+        """
+        if self._client is not None:
+            return
+
+        if not OPENAI_API_KEY:
+            raise RuntimeError("CRITICAL: OPENAI_API_KEY chưa được thiết lập trong config")
+
+        try:
+            self._client = AsyncOpenAI(
+                api_key=OPENAI_API_KEY,
+                # organization=os.getenv("OPENAI_ORG"), # Uncomment if needed
+            )
+            logger.info("✅ OpenAI Client initialized successfully")
+        except Exception as e:
+            logger.error(f"❌ Failed to initialize OpenAI Client: {e}")
+            raise
+
+    def get_client(self) -> AsyncOpenAI:
+        """
+        Lấy client instance (Lazy loading nếu chưa init).
+        """
+        if self._client is None:
+            logger.debug("⚠️ OpenAI Client accessed before init, triggering lazy initialization...")
+            self.initialize()
+
+        if self._client is None:  # Should not happen if init raises error
+            raise RuntimeError("OpenAI Client initialization failed")
+
+        return self._client
+
+
+# --- Singleton Instance & Public API ---
+
+_manager = OpenAIClientManager()
+
+# Alias functions cho backward compatibility
+init_openai_client = _manager.initialize
+get_openai_client = _manager.get_client

backend/datadb/add_context_to_sizes.py

+"""
+Script tự động thêm context (tên bảng + subsection) vào tất cả size entries trong tonghop.txt
+Ví dụ: "Size 92 (2Y):" -> "Size 92 (2Y) - BẢNG SIZE CHUNG CHO UNISEX - TRẺ EM (Dải size lẻ):"
+"""
+
+import re
+
+
+def add_context_to_sizes(input_file, output_file):
+    with open(input_file, encoding="utf-8") as f:
+        lines = f.readlines()
+
+    result = []
+    current_table = None  # Tên bảng hiện tại
+    current_subsection = None  # Subsection hiện tại (Dải size lẻ, chẵn...)
+
+    for line in lines:
+        stripped = line.strip()
+
+        # Phát hiện header bảng (bắt đầu bằng BẢNG hoặc QUẦN)
+        if stripped.startswith("BẢNG SIZE") or stripped.startswith("QUẦN"):
+            current_table = stripped
+            current_subsection = None  # Reset subsection khi sang bảng mới
+            result.append(line)
+            continue
+
+        # Phát hiện subsection (Dải size lẻ, Dải size chẵn)
+        if "Dải size" in stripped or stripped.startswith("Dải"):
+            current_subsection = stripped.rstrip(":")
+            result.append(line)
+            continue
+
+        # Phát hiện dòng Size (bắt mọi pattern: Size XS:, Size 92 (2Y):, Size 26 (XS):)
+        size_match = re.match(r"^(Size\s+[A-Z0-9]+(?:\s*\([^)]+\))?):(.*)$", stripped)
+        if size_match and current_table:
+            size_part = size_match.group(1)  # "Size 92 (2Y)" hoặc "Size XS"
+            rest = size_match.group(2)  # Phần còn lại sau dấu :
+
+            # Xây dựng context
+            context_parts = [current_table]
+            if current_subsection:
+                context_parts.append(f"({current_subsection})")
+
+            context = " - ".join(context_parts)
+
+            # Tạo dòng mới với context
+            new_line = f"{size_part} - {context}:{rest}\n"
+            result.append(new_line)
+            continue
+
+        # Giữ nguyên các dòng khác
+        result.append(line)
+
+    # Ghi file output
+    with open(output_file, "w", encoding="utf-8") as f:
+        f.writelines(result)
+
+    print("✅ Đã thêm context vào tất cả size entries!")
+    print(f"📝 File output: {output_file}")
+
+
+if __name__ == "__main__":
+    input_path = r"d:\cnf\chatbot_canifa\backend\datadb\tonghop.txt"
+    output_path = r"d:\cnf\chatbot_canifa\backend\datadb\tonghop_with_context.txt"
+
+    add_context_to_sizes(input_path, output_path)
+
+    print("\n🔍 Preview 10 dòng đầu của file mới:")
+    with open(output_path, encoding="utf-8") as f:
+        for i, line in enumerate(f):
+            if i >= 1160 and i < 1170:  # Vùng có size entries
+                print(line.rstrip())

backend/docker-compose.yml

+services:
+  # --- Backend Service ---
+  backend:
+    build: .
+    container_name: canifa_backend
+    env_file: .env
+    ports:
+      - "5000:5000"
+    volumes:
+      - .:/app
+    environment:
+      - PORT=5000
+    restart: unless-stopped
+    deploy:
+      resources:
+        limits:
+          memory: 8g
+    logging:
+      driver: "json-file"
+      options:
+        tag: "{{.Name}}"
+