Skip to content

C
chatbot canifa
Commit 6ce1554b authored by Vũ Hoàng Anh's avatar Vũ Hoàng Anh
Browse files
Options

refactor: clean up controller logs + improve stock tool prompt

parent 178108a8
Hide whitespace changes
Inline Side-by-side
Showing with 1525 additions and 407 deletions
backend/agent/controller.py
View file @ 6ce1554b
......@@ -13,10 +13,11 @@ import uuid
from fastapi import BackgroundTasks
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from langfuse import get_client as get_langfuse, observe, propagate_attributes
from langfuse import Langfuse, get_client as get_langfuse
from common.cache import redis_cache
from common.conversation_manager import get_conversation_manager
from common.langfuse_client import get_callback_handler
from config import DEFAULT_MODEL, REDIS_CACHE_TURN_ON
from .controller_helpers import (
......@@ -31,7 +32,7 @@ from .streaming_callback import ProductIDStreamingCallback
logger = logging.getLogger(__name__)
@observe(name="chat-request")
async def chat_controller(
query: str,
identity_key: str,
......@@ -59,22 +60,19 @@ async def chat_controller(
is_authenticated: True if user logged in, False if guest
"""
logger.info("chat_controller start: model=%s, identity_key=%s, auth=%s", model_name, identity_key, is_authenticated)
if images:
logger.info("📸 [CONTROLLER] Received %d image(s), sizes: %s", len(images), [len(img) for img in images])
else:
logger.debug("📸 [CONTROLLER] No images in request")
logger.info("chat_controller start: model=%s, key=%s", model_name, identity_key)
# ====================== CACHE LAYER ======================
if REDIS_CACHE_TURN_ON:
cached_response = await redis_cache.get_response(user_id=identity_key, query=query)
if cached_response:
logger.info(f"⚡ CACHE HIT for identity_key={identity_key}")
logger.info("CACHE HIT key=%s", identity_key)
cached_insight = None
if identity_key:
cached_insight = await load_user_insight_from_redis(identity_key)
if cached_insight:
cached_response = {**cached_response, "user_insight": cached_insight}
memory = await get_conversation_manager()
background_tasks.add_task(
handle_post_chat_async,
......@@ -112,63 +110,24 @@ async def chat_controller(
}
run_id = str(uuid.uuid4())
trace_id = Langfuse.create_trace_id()
session_id = f"{identity_key}-{run_id[:8]}"
# Set Langfuse trace metadata (nests under @observe parent)
langfuse = get_langfuse()
langfuse.update_current_trace(
user_id=identity_key,
session_id=session_id,
tags=["chatbot", "user:authenticated" if is_authenticated else "user:anonymous"],
metadata={
"device_id": device_id,
"customer_id": identity_key if is_authenticated else None,
"model": model_name,
},
)
# Get handler from context → nested under same trace as @observe
try:
langfuse_handler = langfuse.get_current_langchain_handler()
except Exception:
langfuse_handler = None
logger.warning("⚠️ Could not get Langfuse langchain handler")
tags = ["chatbot", "user:authenticated" if is_authenticated else "user:anonymous"]
streaming_callback = ProductIDStreamingCallback()
exec_config = RunnableConfig(
configurable={
"user_id": identity_key,
"transient_images": images or [],
"run_id": run_id,
},
run_id=run_id,
metadata={
"session_id": session_id,
"device_id": device_id,
},
callbacks=[langfuse_handler, streaming_callback] if langfuse_handler else [streaming_callback],
)
logger.info("🌊 Starting LLM streaming...")
event_count = 0
start_time = time.time()
event_count = 0
ai_text_response = ""
final_product_ids = []
all_accumulated_messages = []
seen_message_ids = set()
async def consume_events():
nonlocal event_count, ai_text_response, final_product_ids, all_accumulated_messages, seen_message_ids
nonlocal event_count, all_accumulated_messages, seen_message_ids
async for event in graph.astream(initial_state, config=exec_config):
event_count += 1
elapsed = time.time() - start_time
logger.info(f"📦 Event #{event_count} at t={elapsed:.2f}s | Keys: {list(event.keys())}")
# Accumulate messages from ALL events (deduplicate by ID)
for node_name, node_state in event.items():
for _node, node_state in event.items():
if isinstance(node_state, dict) and "messages" in node_state:
for msg in node_state["messages"]:
msg_id = id(msg)
......@@ -178,154 +137,175 @@ async def chat_controller(
early_response = False
stream_task = asyncio.create_task(consume_events())
# --- Langfuse: wrap everything in a named trace with custom trace_id ---
langfuse = get_langfuse()
observation_ctx = (
langfuse.start_as_current_observation(
as_type="span",
name="chat-request",
trace_context={"trace_id": trace_id},
)
if langfuse
else None
)
if return_user_insight:
# Dev mode: Đợi stream xong để có đầy đủ user_insight
logger.info("🔍 [DEV MODE] Waiting for full stream...")
await stream_task
ai_text_response = streaming_callback.ai_response_text
final_product_ids = streaming_callback.product_skus
logger.info("✅ Stream completed in dev mode")
else:
# Prod mode: Return ngay khi có product_ids
wait_task = asyncio.create_task(streaming_callback.product_found_event.wait())
done, pending = await asyncio.wait(
[stream_task, wait_task],
return_when=asyncio.FIRST_COMPLETED,
timeout=30.0,
if observation_ctx:
span = observation_ctx.__enter__()
span.update_trace(
name="chat-request",
user_id=identity_key,
session_id=session_id,
tags=tags,
input={"query": query, "user_id": identity_key},
metadata={
"device_id": device_id,
"customer_id": identity_key if is_authenticated else None,
"model": model_name,
},
)
# bắn trả về ngay khi có product_ids hoặc timeout
ai_text_response = streaming_callback.ai_response_text
final_product_ids = streaming_callback.product_skus
# CallbackHandler MUST be created INSIDE the observation context
# so it inherits the trace_id and nests under "chat-request"
langfuse_handler = get_callback_handler()
exec_config = RunnableConfig(
configurable={
"user_id": identity_key,
"transient_images": images or [],
"run_id": run_id,
"trace_id": trace_id,
},
run_id=run_id,
metadata={
"trace_id": trace_id,
"session_id": session_id,
"device_id": device_id,
"customer_id": identity_key if is_authenticated else None,
},
callbacks=[langfuse_handler, streaming_callback] if langfuse_handler else [streaming_callback],
)
try:
stream_task = asyncio.create_task(consume_events())
elapsed = time.time() - start_time
if streaming_callback.product_ids_found:
logger.info(f"⚡ Response ready at t={elapsed:.2f}s (early return)")
early_response = True
if return_user_insight:
await stream_task
ai_text_response = streaming_callback.ai_response_text
final_product_ids = streaming_callback.product_skus
else:
logger.info(f"✅ Stream completed at t={elapsed:.2f}s (no products)")
wait_task = asyncio.create_task(streaming_callback.product_found_event.wait())
done, pending = await asyncio.wait(
[stream_task, wait_task],
return_when=asyncio.FIRST_COMPLETED,
timeout=30.0,
)
ai_text_response = streaming_callback.ai_response_text
final_product_ids = streaming_callback.product_skus
# Cleanup (never cancel stream_task; we still need it)
if wait_task in pending:
wait_task.cancel()
if streaming_callback.product_ids_found:
early_response = True
if wait_task in pending:
wait_task.cancel()
finally:
# Update trace output before closing
if observation_ctx:
try:
elapsed = time.time() - start_time
span.update_trace(
output={
"ai_response": (ai_text_response or "")[:500],
"product_count": len(final_product_ids) if final_product_ids else 0,
"elapsed_s": round(elapsed, 2),
"early_return": early_response,
},
)
except Exception:
pass
try:
observation_ctx.__exit__(None, None, None)
except Exception:
pass
try:
# Ensure stream completes to get all tool messages (skip if early response)
# Ensure stream completes for tool messages
if not early_response and not stream_task.done():
logger.info("⏳ Waiting for stream to complete to extract tool products...")
try:
await stream_task
except asyncio.CancelledError:
logger.warning("⚠️ stream_task was cancelled; skip tool product extraction")
logger.warning("stream_task cancelled")
# Early return path: try to extract ai_response directly from streaming content
# Extract ai_response from streaming content (fallbacks)
if early_response and not ai_text_response:
raw_content = streaming_callback.accumulated_content
if raw_content:
# Try JSON parse first
try:
ai_json = json.loads(raw_content)
if isinstance(ai_json, dict):
ai_text_response = ai_json.get("ai_response", ai_text_response)
if not final_product_ids and isinstance(ai_json.get("product_ids"), list):
final_product_ids = [str(s) for s in ai_json.get("product_ids", [])]
logger.info("ℹ️ Extracted ai_response from streaming JSON content")
final_product_ids = [str(s) for s in ai_json["product_ids"]]
except json.JSONDecodeError:
# Fallback regex: extract ai_response regardless of field order
ai_match = re.search(r'"ai_response"\s*:\s*"((?:[^"\\]|\\.)*)"', raw_content)
if ai_match:
ai_text_response = ai_match.group(1).replace('\\"', '"').replace("\\n", "\n")
logger.info("ℹ️ Extracted ai_response via regex from streaming content")
# Fallback: If ai_response is empty, take last AI message from accumulated events
if not ai_text_response and all_accumulated_messages:
for msg in reversed(all_accumulated_messages):
if isinstance(msg, AIMessage) and msg.content:
ai_text_response = msg.content
logger.info("ℹ️ Filled ai_response from last AIMessage in accumulated events")
break
# If ai_response looks like JSON, extract ai_response field directly
# Parse JSON-wrapped ai_response
if ai_text_response and ai_text_response.lstrip().startswith("{"):
try:
ai_json = json.loads(ai_text_response)
if isinstance(ai_json, dict):
ai_text_response = ai_json.get("ai_response", ai_text_response)
if not final_product_ids and isinstance(ai_json.get("product_ids"), list):
final_product_ids = [str(s) for s in ai_json.get("product_ids", [])]
logger.info("ℹ️ Extracted ai_response from JSON content")
final_product_ids = [str(s) for s in ai_json["product_ids"]]
except json.JSONDecodeError:
logger.warning("⚠️ ai_response looked like JSON but failed to parse")
pass
# ✅ Extract products
# PRIORITY: Use tool results from accumulated_messages (has correct color)
# FALLBACK: Only use fetch_products_by_skus if no tool results available
enriched_products = [] # Initialize before if block
# Extract & filter products
enriched_products = []
if all_accumulated_messages:
all_extracted_products = extract_product_ids(all_accumulated_messages)
logger.info(
f"✅ Extracted {len(all_extracted_products)} products from {len(all_accumulated_messages)} accumulated messages"
)
# 🎯 FILTER: Only return products that AI mentioned in product_ids
all_extracted = extract_product_ids(all_accumulated_messages)
if final_product_ids:
# AI mentioned specific SKUs → Filter to match only those
mentioned_skus = set(final_product_ids)
enriched_products = [p for p in all_extracted_products if p.get("sku") in mentioned_skus]
logger.warning(
f"🎯 AI mentioned {len(mentioned_skus)} SKUs → Filtered {len(all_extracted_products)} down to {len(enriched_products)} products"
)
mentioned = set(final_product_ids)
enriched_products = [p for p in all_extracted if p.get("sku") in mentioned]
else:
# AI didn't mention specific SKUs → Return all extracted products
enriched_products = all_extracted_products
logger.info("ℹ️ No specific SKUs mentioned by AI, returning all extracted products")
else:
logger.warning("⚠️ No accumulated messages found - returning empty product list")
if not enriched_products:
logger.warning("⚠️ No products extracted from tool messages - returning empty list")
enriched_products = all_extracted
except Exception as e:
logger.error(f"❌ Error extracting products from tool messages: {e}")
logger.error("Error extracting products: %s", e)
enriched_products = []
user_insight_dict = None
if return_user_insight:
callback_accumulated_content = streaming_callback.accumulated_content
if callback_accumulated_content and identity_key:
logger.info("🔍 [DEV] Extracting user_insight...")
user_insight_dict = await extract_and_save_user_insight(callback_accumulated_content, identity_key)
content = streaming_callback.accumulated_content
if content and identity_key:
user_insight_dict = await extract_and_save_user_insight(content, identity_key)
elif background_tasks:
# Prod/Dev fast mode: để stream chạy xong rồi mới extract trong background
logger.info("💾 [PROD] Scheduling background task for user_insight extraction (post-stream)")
background_tasks.add_task(
finalize_user_insight_after_stream,
stream_task=stream_task,
streaming_callback=streaming_callback,
identity_key=identity_key,
trace_id=langfuse.get_current_trace_id() if langfuse else run_id,
trace_id=trace_id,
)
# ====================== BUILD RESPONSE ======================
response_payload = {
"ai_response": ai_text_response,
"product_ids": enriched_products, # ⚡ Full product objects from tool (name, price, sale_price, url, thumbnail)
"product_ids": enriched_products,
}
if user_insight_dict is not None:
response_payload["user_insight"] = user_insight_dict
# ====================== SAVE TO CACHE ======================
if REDIS_CACHE_TURN_ON:
await redis_cache.set_response(user_id=identity_key, query=query, response_data=response_payload, ttl=300)
logger.debug(f"💾 Cached response for identity_key={identity_key}")
# Save to History (Background)
background_tasks.add_task(
handle_post_chat_async,
memory=memory,
......@@ -333,6 +313,4 @@ async def chat_controller(
human_query=query,
ai_response=response_payload,
)
logger.info("chat_controller finished: cached=False")
return {**response_payload, "cached": False}
backend/agent/controller_helpers.py
View file @ 6ce1554b
......@@ -109,39 +109,29 @@ async def finalize_user_insight_after_stream(
stream_task,
streaming_callback,
identity_key: str,
trace_id: str,
trace_id: str = "",
) -> None:
"""
Background task: Wait for stream to complete, then extract and save user_insight.
Links to main trace using trace_context for proper Langfuse tracing.
Args:
stream_task: The asyncio task for graph streaming
streaming_callback: ProductIDStreamingCallback instance
identity_key: User/device identifier
trace_id: Langfuse trace ID to link background task
NOTE: Runs outside Langfuse trace context — no need to log to Langfuse.
"""
# Clear Langfuse observation context to avoid auto-tracing in background
try:
from langfuse._context import _observation_stack_context
_observation_stack_context.set([])
except Exception:
pass # Langfuse version may not have this; safe to ignore
try:
from langfuse import get_client
langfuse = get_client()
# Link background task to same trace using trace_context
with langfuse.start_as_current_observation(
name="extract-user-insight-background", trace_context={"trace_id": trace_id}
) as span:
# Wait for stream to complete
await stream_task
# Extract user insight from accumulated content
full_content = streaming_callback.accumulated_content
if full_content and identity_key:
span.update(input={"identity_key": identity_key, "content_length": len(full_content)})
result = await extract_and_save_user_insight(full_content, identity_key)
span.update(output=result)
else:
logger.warning("⚠️ [Background] No content to extract user_insight from")
span.update(output={"status": "no_content"})
# Wait for stream to complete
await stream_task
# Extract user insight from accumulated content
full_content = streaming_callback.accumulated_content
if full_content and identity_key:
await extract_and_save_user_insight(full_content, identity_key)
else:
logger.warning("⚠️ [Background] No content to extract user_insight from")
except Exception as exc:
logger.error(f"❌ [Background] user_insight finalize failed: {exc}")
......
backend/agent/graph.py
View file @ 6ce1554b
......@@ -21,7 +21,7 @@ from langgraph.types import CachePolicy
from common.llm_factory import create_llm
from .models import AgentConfig, AgentState, get_config
from .prompt import get_last_modified, get_system_prompt, get_system_prompt_template
from .prompt_utils import get_system_prompt_template
from .tools.get_tools import get_all_tools, get_collection_tools
logger = logging.getLogger(__name__)
......@@ -48,11 +48,13 @@ class CANIFAGraph:
self.retrieval_tools = self.all_tools
self.llm_with_tools = self.llm.bind_tools(self.all_tools, strict=True)
# Lưu template với {date_str} placeholder → inject dynamic mỗi request
self.system_prompt_template = get_system_prompt_template()
self.prompt_template = ChatPromptTemplate.from_messages(
self.cache = InMemoryCache()
def _build_chain(self, system_prompt_template: str):
"""Build chain with dynamic system prompt (fetched from Langfuse per request)."""
prompt_template = ChatPromptTemplate.from_messages(
[
("system", self.system_prompt_template),
("system", system_prompt_template),
(
"system",
"===== USER INSIGHT (TỪ TURN TRƯỚC) =====\n⚡ BẮT BUỘC: Đọc [NEXT] bên dưới và THỰC HIỆN chiến lược đã lên kế hoạch!\n\n{user_insight}\n=====================================",
......@@ -64,9 +66,7 @@ class CANIFAGraph:
MessagesPlaceholder(variable_name="messages"),
]
)
self.chain = self.prompt_template | self.llm_with_tools
self.cache = InMemoryCache()
return prompt_template | self.llm_with_tools
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."""
......@@ -104,10 +104,12 @@ class CANIFAGraph:
or "⚠️ TRẠNG THÁI KHỞI TẠO: Chưa có User Insight từ lịch sử. Hãy bắt đầu thu thập thông tin mới (Nếu thiếu thông tin thì ghi 'Chưa rõ')."
)
# Inject date_str ĐỘNG mỗi request (không cache từ __init__)
# Fetch prompt from Langfuse DYNAMICALLY per request (cache_ttl=0 → 30ms)
current_date_str = datetime.now().strftime("%d/%m/%Y")
system_prompt_template = get_system_prompt_template()
chain = self._build_chain(system_prompt_template)
response = await self.chain.ainvoke(
response = await chain.ainvoke(
{
"date_str": current_date_str,
"history": history,
......@@ -181,34 +183,23 @@ def build_graph(config: AgentConfig | None = None, llm: BaseChatModel | None = N
def get_graph_manager(
config: AgentConfig | None = None, llm: BaseChatModel | None = None, tools: list | None = None
) -> CANIFAGraph:
"""Get CANIFAGraph instance (Auto-rebuild if model config changes OR prompt version changed)."""
current_prompt_version = get_last_modified()
"""Get CANIFAGraph instance (Auto-rebuild if model config changes).
Prompt is now fetched dynamically per request from Langfuse,
so no need to rebuild graph when prompt changes.
"""
# 1. New Instance if Empty
if _instance[0] is None:
_instance[0] = CANIFAGraph(config, llm, tools)
_instance[0].prompt_version = current_prompt_version
logger.info(f"✨ Graph Created: {_instance[0].config.model_name}, prompt_version={current_prompt_version}")
logger.info(f"✨ Graph Created: {_instance[0].config.model_name} (prompts from Langfuse)")
return _instance[0]
# 2. Check for Config Changes (Model Switch OR Prompt Version Change)
# 2. Check for Model Config Changes only
is_model_changed = config and config.model_name != _instance[0].config.model_name
is_prompt_changed = current_prompt_version != getattr(_instance[0], "prompt_version", 0)
if is_model_changed or is_prompt_changed:
change_reason = []
if is_model_changed:
change_reason.append(f"Model ({_instance[0].config.model_name}->{config.model_name})")
if is_prompt_changed:
change_reason.append(
f"Prompt Version ({getattr(_instance[0], 'prompt_version', 0)}->{current_prompt_version})"
)
logger.info(f"🔄 Rebuilding Graph due to: {', '.join(change_reason)}")
if is_model_changed:
logger.info(f"🔄 Rebuilding Graph: Model ({_instance[0].config.model_name}->{config.model_name})")
_instance[0] = CANIFAGraph(config, llm, tools)
_instance[0].prompt_version = current_prompt_version
return _instance[0]
return _instance[0]
......
backend/agent/prompt.py deleted 100644 → 0
View file @ 178108a8
"""
CiCi Fashion Consultant - System Prompt
Tư vấn thời trang CANIFA chuyên nghiệp
Version 3.4 - Simplified Summary History
Last updated: 2026-01-29 11:27
"""
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.
"""
def get_system_prompt_template() -> str:
"""
Trả về system prompt template CHƯA replace {date_str}.
Dùng cho ChatPromptTemplate để inject ngày động mỗi request.
Returns:
str: System prompt template với placeholder {date_str}
"""
try:
if os.path.exists(PROMPT_FILE_PATH):
with open(PROMPT_FILE_PATH, "r", encoding="utf-8") as f:
return f.read()
except Exception as e:
print(f"Error reading system prompt file: {e}")
# Fallback default prompt if file error
return """# 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.
"""
def get_last_modified() -> float:
"""Trả về timestamp lần sửa cuối cùng của file system_prompt.txt."""
try:
if os.path.exists(PROMPT_FILE_PATH):
return os.path.getmtime(PROMPT_FILE_PATH)
except Exception:
pass
return 0.0
\ No newline at end of file
backend/agent/prompt_utils.py
View file @ 6ce1554b
"""
Prompt Utilities — ALL prompts from Langfuse.
System prompt + Tool prompts, single source of truth.
"""
import os
import logging
from datetime import datetime
from langfuse import Langfuse
logger = logging.getLogger(__name__)
# Directory name for tool prompts
PROMPTS_DIR_NAME = "tool_prompts"
PROMPTS_DIR = os.path.join(os.path.dirname(__file__), PROMPTS_DIR_NAME)
LANGFUSE_SYSTEM_PROMPT_NAME = "canifa-stylist-system-prompt"
LANGFUSE_TOOL_PROMPT_MAP = {
"brand_knowledge_tool": "canifa-tool-brand-knowledge",
"check_is_stock": "canifa-tool-check-stock",
"data_retrieval_tool": "canifa-tool-data-retrieval",
"promotion_canifa_tool": "canifa-tool-promotion",
}
_langfuse_client: Langfuse | None = None
def _get_langfuse() -> Langfuse:
global _langfuse_client
if _langfuse_client is None:
_langfuse_client = Langfuse()
return _langfuse_client
def get_system_prompt() -> str:
"""System prompt với ngày hiện tại đã inject."""
lf = _get_langfuse()
prompt = lf.get_prompt(LANGFUSE_SYSTEM_PROMPT_NAME, label="production", cache_ttl_seconds=0)
return prompt.compile(date_str=datetime.now().strftime("%d/%m/%Y"))
def get_system_prompt_template() -> str:
"""Template chưa replace date_str — dùng cho ChatPromptTemplate."""
lf = _get_langfuse()
prompt = lf.get_prompt(LANGFUSE_SYSTEM_PROMPT_NAME, label="production", cache_ttl_seconds=0)
return prompt.prompt.replace("{{date_str}}", "{date_str}")
def get_tool_prompt_path(filename: str) -> str:
"""Get absolute path for a tool prompt file."""
if not filename.endswith(".txt"):
filename += ".txt"
return os.path.join(PROMPTS_DIR, filename)
def read_tool_prompt(filename: str, default_prompt: str = "") -> str:
"""
Read tool prompt from file.
Returns default_prompt if file not found or empty.
"""
file_path = get_tool_prompt_path(filename)
try:
if os.path.exists(file_path):
with open(file_path, "r", encoding="utf-8") as f:
content = f.read().strip()
if content:
return content
except Exception as e:
logger.error(f"Error reading tool prompt {filename}: {e}")
return default_prompt
"""Read tool prompt from Langfuse."""
name_key = filename.replace(".txt", "")
langfuse_name = LANGFUSE_TOOL_PROMPT_MAP.get(name_key)
if not langfuse_name:
logger.warning(f"⚠️ No Langfuse mapping for tool prompt '{name_key}'")
return default_prompt
lf = _get_langfuse()
prompt = lf.get_prompt(langfuse_name, label="production", cache_ttl_seconds=0)
return prompt.prompt
def write_tool_prompt(filename: str, content: str) -> bool:
"""Write content to tool prompt file."""
file_path = get_tool_prompt_path(filename)
try:
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w", encoding="utf-8") as f:
f.write(content)
return True
except Exception as e:
logger.error(f"Error writing tool prompt {filename}: {e}")
"""Push tool prompt to Langfuse as new version."""
name_key = filename.replace(".txt", "")
langfuse_name = LANGFUSE_TOOL_PROMPT_MAP.get(name_key)
if not langfuse_name:
logger.error(f"No Langfuse mapping for '{name_key}'")
return False
lf = _get_langfuse()
lf.create_prompt(
name=langfuse_name,
prompt=content,
labels=["production"],
tags=["canifa", "tool-prompt"],
type="text",
)
logger.info(f"✅ Tool prompt '{name_key}' pushed to Langfuse")
return True
def list_tool_prompts() -> list[str]:
"""List all available tool prompt files."""
try:
if not os.path.exists(PROMPTS_DIR):
return []
files = [f for f in os.listdir(PROMPTS_DIR) if f.endswith(".txt")]
return sorted(files)
except Exception as e:
logger.error(f"Error listing tool prompts: {e}")
return []
"""List available tool prompt names."""
return sorted(LANGFUSE_TOOL_PROMPT_MAP.keys())
backend/agent/tool_prompts/check_is_stock.txt
View file @ 6ce1554b
......@@ -2,6 +2,7 @@ Công cụ KIỂM TRA TỒN KHO sản phẩm CANIFA theo mã sản phẩm.
KHI NÀO GỌI TOOL NÀY:
- Khách hỏi "còn hàng không?", "còn size không?", "check tồn kho"
- Khách hỏi "có những size nào?", "còn những size nào?", "size chart", "số lượng từng size"
- Khách hỏi về MÃ SẢN PHẨM CỤ THỂ kèm từ khóa tồn kho (vd: "8TS24W001 còn size L không?")
- Khách muốn biết số lượng tồn kho của một hoặc nhiều mã sản phẩm
......@@ -30,7 +31,6 @@ CASE 3: KIỂM TRA MÃ KÈM SIZE
User: "Mã 6ST25W005-SE091 còn size M và L không?"
-> Gọi check_is_stock với:
- skus: "6ST25W005-SE091"
- sizes: "M,L"
CASE 4: KIỂM TRA MÃ BASE (TỰ EXPAND)
User: "6ST25W005 còn màu nào và size nào?"
......@@ -38,8 +38,23 @@ User: "6ST25W005 còn màu nào và size nào?"
- skus: "6ST25W005"
(Tool sẽ tự động expand ra tất cả các biến thể từ DB)
CÁCH ĐỌC KẾT QUẢ:
CASE 5: HỎI CÓ NHỮNG SIZE NÀO / CÒN SIZE NÀO
User: "Áo này có những size nào?" hoặc "Sản phẩm này còn size gì?"
-> Gọi check_is_stock với mã sản phẩm (lấy từ context cuộc hội thoại)
- skus: "6ST25W005-SE091"
CÁCH ĐỌC VÀ TRÌNH BÀY KẾT QUẢ:
- stock_responses: Danh sách tồn kho từng SKU
- is_in_stock: true/false - còn hàng hay không
- qty: số lượng tồn kho
- Nếu hết hàng -> Gợi ý size/màu khác còn hàng
CÁCH TRÌNH BÀY CHO KHÁCH:
1. Liệt kê RÕ RÀNG từng size kèm tình trạng:
- Size S: Còn hàng (12 sản phẩm)
- Size M: Còn hàng (8 sản phẩm)
- Size L: Hết hàng ❌
- Size XL: Còn hàng (3 sản phẩm)
2. Nếu khách hỏi "có những size nào" → Liệt kê TẤT CẢ size có trong kết quả, kèm số lượng
3. Nếu khách hỏi "còn size nào" → Chỉ liệt kê size CÒN HÀNG (qty > 0)
4. Nếu hết hàng size khách muốn → Gợi ý size/màu khác còn hàng
5. Nếu sản phẩm hoàn toàn hết hàng → Thông báo rõ và gợi ý sản phẩm tương tự
backend/agent/tools/data_retrieval_tool.py
View file @ 6ce1554b
"""
CANIFA Data Retrieval Tool - Tối giản cho Agentic Workflow.
Hỗ trợ Hybrid Search: Semantic (Vector) + Metadata Filter.
Tracing:
- LangChain CallbackHandler auto-traces tool call (name, input, output)
- Manual span 'db-query-search' in _execute_single_search for SQL timing & product count
(nested under parent LangChain trace via start_as_current_observation)
"""
import asyncio
......@@ -8,7 +13,8 @@ import json
import logging
import time
from langfuse import observe
from langfuse import get_client as get_langfuse
from langchain_core.tools import tool
from pydantic import BaseModel, Field
......@@ -103,12 +109,12 @@ class MultiSearchParams(BaseModel):
searches: list[SearchItem] = Field(description="Danh sách các truy vấn tìm kiếm")
@observe(name="db-query-search")
async def _execute_single_search(
db, item: SearchItem, query_vector: list[float] | None = None
) -> tuple[list[dict], dict]:
"""
Thực thi một search query đơn lẻ (Async).
Creates a manual 'db-query-search' span under the current Langfuse context.
Returns:
Tuple of (products, filter_info)
......@@ -130,14 +136,15 @@ async def _execute_single_search(
logger.debug("SQL built, length=%s, build_time_ms=%.2f", len(sql), query_build_time)
if not sql:
return []
return [], {"fallback_used": False}
# Timer: execute DB query
db_start = time.time()
products = await db.execute_query_async(sql, params=params)
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",
"🔍 _execute_single_search done: query=%r, products=%s, build_ms=%.2f, db_ms=%.2f, total_ms=%.2f",
short_query,
len(products),
query_build_time,
db_time,
......@@ -159,26 +166,44 @@ async def _execute_single_search(
(first_p.get("description_text") or "")[:100],
)
from langfuse import get_client as get_langfuse
# ✅ Nest 'db-query-search' span dưới chat-request trace hiện tại trên Langfuse
try:
lf = get_langfuse()
lf.update_current_observation(
input={
"description": item.description[:200] if item.description else None,
"product_name": item.product_name,
"gender": item.gender_by_product,
"age": item.age_by_product,
"product_line_vn": item.product_line_vn,
"magento_ref_code": item.magento_ref_code,
},
output={
"raw_count": len(products),
"build_ms": round(query_build_time, 2),
"db_ms": round(db_time, 2),
},
)
except Exception:
pass # Don't break if Langfuse not available
if lf:
with lf.start_as_current_observation(
as_type="span",
name="db-query-search",
) as span:
span.update(
input={
"description": item.description[:200] if item.description else None,
"product_name": item.product_name,
"gender": item.gender_by_product,
"age": item.age_by_product,
"color": item.master_color,
"price_range": f"{item.price_min or ''}-{item.price_max or ''}",
"product_line_vn": item.product_line_vn,
"magento_ref_code": item.magento_ref_code,
"discovery_mode": item.discovery_mode,
},
output={
"product_count": len(products),
"build_ms": round(query_build_time, 2),
"db_ms": round(db_time, 2),
"total_ms": round(query_build_time + db_time, 2),
"sample_products": [
{
"name": p.get("product_name", "?"),
"code": p.get("magento_ref_code", "?"),
"price": float(p["sale_price"]) if p.get("sale_price") else None,
}
for p in products[:3]
] if products else [],
},
metadata={"sql_length": len(sql)},
)
except Exception as trace_err:
logger.warning("⚠️ Langfuse db-query-search span failed: %s", trace_err, exc_info=True)
return format_product_results(products), {"fallback_used": False}
except Exception as e:
......@@ -192,8 +217,26 @@ async def data_retrieval_tool(searches: list[SearchItem]) -> str:
Công cụ tìm kiếm sản phẩm CANIFA.
Hỗ trợ tìm kiếm Semantic và lọc theo Metadata.
"""
tool_start = time.time()
logger.info("🔧 data_retrieval_tool called with %d items", len(searches))
# Log search params cho debugging
for i, s in enumerate(searches):
logger.info(
"🔧 Search[%d]: desc=%r, name=%r, gender=%s, age=%s, color=%s, "
"price=%s-%s, line=%s, code=%s, mode=%s",
i,
(s.description[:80] + "...") if s.description and len(s.description) > 80 else s.description,
s.product_name,
s.gender_by_product,
s.age_by_product,
s.master_color,
s.price_min, s.price_max,
s.product_line_vn,
s.magento_ref_code,
s.discovery_mode,
)
# Get DB Connection
db = get_db_connection()
if not db:
......@@ -213,8 +256,6 @@ async def data_retrieval_tool(searches: list[SearchItem]) -> str:
if filter_info:
all_filter_infos.append(filter_info)
# Aggregate filter info from first result for simplicity in response
final_info = all_filter_infos[0] if all_filter_infos else {}
......@@ -237,16 +278,18 @@ async def data_retrieval_tool(searches: list[SearchItem]) -> str:
"filter_info": final_info,
}
total_ms = (time.time() - tool_start) * 1000
logger.info(
"🎁 Final result: %d products. Fallback used: %s",
"🎁 Final result: %d products, total_ms=%.2f. Fallback used: %s",
len(combined_results),
total_ms,
final_info.get("fallback_used", False),
)
return json.dumps(output, ensure_ascii=False, default=str)
# Load dynamic docstring
# Load dynamic docstring
dynamic_prompt = read_tool_prompt("data_retrieval_tool")
if dynamic_prompt:
......
backend/agent/tools/product_search_helpers.py
View file @ 6ce1554b
......@@ -2,7 +2,7 @@ import logging
import os
import time
from langfuse import observe
# Note: tracing handled by parent data_retrieval_tool via context manager
from common.embedding_service import create_embedding_async
......@@ -103,7 +103,6 @@ def _get_metadata_clauses(params, sql_params: list) -> list[str]:
return clauses
@observe(name="build-starrocks-query")
async def build_starrocks_query(params, query_vector: list[float] | None = None) -> tuple[str, list]:
"""
Build SQL query với Parameterized Query để tránh SQL Injection.
......
backend/api/prompt_route.py
View file @ 6ce1554b
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel
import os
import re
from agent.graph import reset_graph
from common.cache import bump_prompt_version
router = APIRouter()
from agent.prompt_utils import get_system_prompt_template, LANGFUSE_SYSTEM_PROMPT_NAME, _get_langfuse
PROMPT_FILE_PATH = os.path.join(os.path.dirname(__file__), "../agent/system_prompt.txt")
router = APIRouter()
# Allowed variables in prompt (single braces OK for these)
ALLOWED_VARIABLES = {"date_str"}
class PromptUpdateRequest(BaseModel):
content: str
def validate_prompt_braces(content: str) -> tuple[bool, list[str]]:
"""
Validate that all braces in prompt are properly escaped.
Returns (is_valid, list of problematic patterns)
"""
# Find all {word} patterns
single_brace_pattern = re.findall(r'\{([^{}]+)\}', content)
# Filter out allowed variables
problematic = [
var for var in single_brace_pattern
if var.strip() not in ALLOWED_VARIABLES and not var.startswith('{')
]
return len(problematic) == 0, problematic
from common.rate_limit import rate_limit_service
@router.get("/api/agent/system-prompt")
async def get_system_prompt_content(request: Request):
"""Get current system prompt content"""
"""Get current system prompt content from Langfuse."""
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"}
return {"status": "success", "content": get_system_prompt_template()}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
......@@ -50,41 +25,24 @@ async def get_system_prompt_content(request: Request):
@router.post("/api/agent/system-prompt")
@rate_limit_service.limiter.limit("10/minute")
async def update_system_prompt_content(request: Request, body: PromptUpdateRequest):
"""Update system prompt content"""
"""Create new system prompt version in Langfuse."""
try:
# Validate braces
is_valid, problematic = validate_prompt_braces(body.content)
if not is_valid:
# Return warning but still allow save
warning = (
f"⚠️ Phát hiện {{...}} chưa escape: {problematic[:3]}... "
f"Nếu đây là JSON, hãy dùng {{{{ }}}} thay vì {{ }}. "
f"Prompt vẫn được lưu nhưng có thể gây lỗi khi chat."
)
else:
warning = None
# 1. Update file
with open(PROMPT_FILE_PATH, "w", encoding="utf-8") as f:
f.write(body.content)
# 2. Bump prompt version in Redis (ALL workers will detect this)
new_version = await bump_prompt_version()
# 3. Reset local worker's Graph Singleton (immediate effect for this worker)
reset_graph()
response = {
"status": "success",
"message": f"System prompt updated. Version: {new_version}. All workers will reload on next request.",
"prompt_version": new_version
lf = _get_langfuse()
content = body.content
if "{date_str}" in content and "{{date_str}}" not in content:
content = content.replace("{date_str}", "{{date_str}}")
prompt = lf.create_prompt(
name=LANGFUSE_SYSTEM_PROMPT_NAME,
prompt=content,
labels=["production"],
tags=["canifa", "system-prompt"],
type="text",
)
return {
"status": "success",
"message": f"System prompt updated. Version: {prompt.version}",
"prompt_version": prompt.version,
}
if warning:
response["warning"] = warning
return response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
backend/api/tool_prompt_route.py
View file @ 6ce1554b
......@@ -23,7 +23,7 @@ async def get_tool_prompts_list(request: Request):
@router.get("/api/agent/tool-prompts/{filename}")
async def get_tool_prompt_content(filename: str, request: Request):
"""Get content of a specific tool prompt file."""
"""Get content of a specific tool prompt (from Langfuse, fallback to file)."""
try:
content = read_tool_prompt(filename)
if not content:
......@@ -38,7 +38,7 @@ async def get_tool_prompt_content(filename: str, request: Request):
@router.post("/api/agent/tool-prompts/{filename}")
async def update_tool_prompt_content(filename: str, request: Request, body: ToolPromptUpdateRequest):
"""Update content of a tool prompt file and reset graph."""
"""Update tool prompt (writes to file + syncs to Langfuse) and reset graph."""
try:
# Ensure filename is safe (basic check)
if ".." in filename or "/" in filename or "\\" in filename:
......@@ -51,7 +51,7 @@ async def update_tool_prompt_content(filename: str, request: Request, body: Tool
# Reset Graph to reload tools with new prompts
reset_graph()
return {"status": "success", "message": f"Tool prompt {filename} updated successfully. Graph reloaded."}
return {"status": "success", "message": f"Tool prompt {filename} updated (file + Langfuse). Graph reloaded."}
except HTTPException:
raise
except Exception as e:
......
backend/tests/test_model.py 0 → 100644
View file @ 6ce1554b
"""Quick test: Gemini 2.0 Flash via Google GenAI."""
from google import genai
API_KEY = "AIzaSyCLAk9LsvS7RtxuXJLf5uqx4sI034Hc3Xg"
client = genai.Client(api_key=API_KEY)
response = client.models.generate_content(
model="gemini-3-flash-preview",
contents="Xin chào! Bạn là ai? Trả lời ngắn gọn bằng tiếng Việt.",
)
print("=" * 60)
print("Model: gemini--flash")
print("=" * 60)
print(response.text)
print("=" * 60)
public/trace-7dc7d62af563d3779549a7765408d8f6.json 0 → 100644
View file @ 6ce1554b
This source diff could not be displayed because it is too large. You can view the blob instead.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment