wangliang/assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加图片搜索功能和 Qwen 模型支持

Browse Source 
图片搜索功能（以图搜图）：
- Chatwoot webhook 检测图片搜索消息 (content_type="search_image")
- 从 content_attributes.url 提取图片 URL
- 调用 Mall API 图片搜索接口 (/mall/api/spu?searchImageUrl=...)
- 支持嵌套和顶层 URL 位置提取
- Product Agent 添加 fast path 直接调用图片搜索工具
- 防止无限循环（使用后清除 context.image_search_url）

Qwen 模型支持：
- 添加 LLM provider 选择（zhipu/qwen）
- 实现 QwenLLMClient 类（基于 DashScope SDK）
- 添加 dashscope>=1.14.0 依赖
- 修复 API key 设置（直接设置 dashscope.api_key）
- 更新 .env.example 和 docker-compose.yml 配置

其他优化：
- 重构 Chatwoot 集成代码（删除冗余）
- 优化 Product Agent prompt
- 增强 Customer Service Agent 多语言支持

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
wangliang
2026-01-27 19:10:06 +08:00
parent 754804219f
commit 965b11316e
12 changed files with 937 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
agent/agents/customer_service.py
View File

@@ -322,16 +322,51 @@ async def customer_service_agent(state: AgentState) -> AgentState:
return state
except json.JSONDecodeError as e:
# JSON parsing failed
# JSON parsing failed - try alternative format: "tool_name\n{args}"
logger.error(
"Failed to parse LLM response as JSON",
error=str(e),
raw_content=content[:500],
conversation_id=state["conversation_id"]
)
# Don't use raw content as response - use fallback instead
state = set_response(state, "抱歉，我无法理解您的请求。请尝试重新表述或联系人工客服。")
return state
# Handle non-JSON format: "tool_name\n{args}"
if '\n' in content and not content.startswith('{'):
lines = content.split('\n', 1)
tool_name = lines[0].strip()
args_json = lines[1].strip() if len(lines) > 1 else '{}'
try:
arguments = json.loads(args_json) if args_json else {}
logger.info(
"Customer service agent calling tool (alternative format)",
tool_name=tool_name,
arguments=arguments,
conversation_id=state["conversation_id"]
)
state = add_tool_call(
state,
tool_name=tool_name,
arguments=arguments,
server="strapi"
)
state["state"] = ConversationState.TOOL_CALLING.value
return state
except json.JSONDecodeError:
# Args parsing also failed
logger.warning(
"Failed to parse tool arguments",
tool_name=tool_name,
args_json=args_json[:200],
conversation_id=state["conversation_id"]
)
state = set_response(state, "抱歉，我无法理解您的请求。请尝试重新表述或联系人工客服。")
return state
else:
# Not a recognized format
state = set_response(state, "抱歉，我无法理解您的请求。请尝试重新表述或联系人工客服。")
return state
except Exception as e:
logger.error("Customer service agent failed", error=str(e), exc_info=True)
@@ -342,11 +377,59 @@ async def customer_service_agent(state: AgentState) -> AgentState:
async def _generate_response_from_results(state: AgentState) -> AgentState:
"""Generate response based on tool results"""
# Build context from tool results
# Build context from tool results - extract only essential info to reduce prompt size
tool_context = []
for result in state["tool_results"]:
if result["success"]:
tool_context.append(f"Tool {result['tool_name']} returned:\n{json.dumps(result['data'], ensure_ascii=False, indent=2)}")
tool_name = result['tool_name']
data = result['data']
# Extract only essential information based on tool type
if tool_name == "get_company_info":
# Extract key contact info only
contact = data.get('contact', {})
emails = contact.get('email', [])
if isinstance(emails, list) and emails:
email_str = ", ".join(emails[:3]) # Max 3 emails
else:
email_str = str(emails) if emails else "N/A"
phones = contact.get('phone', [])
if isinstance(phones, list) and phones:
phone_str = ", ".join(phones[:2]) # Max 2 phones
else:
phone_str = str(phones) if phones else "N/A"
address = contact.get('address', {})
address_str = f"{address.get('city', '')}, {address.get('country', '')}".strip(', ')
summary = f"Contact Information: Emails: {email_str} | Phones: {phone_str} | Address: {address_str} | Working hours: {contact.get('working_hours', 'N/A')}"
tool_context.append(summary)
elif tool_name == "query_faq" or tool_name == "search_knowledge_base":
# Extract FAQ items summary
faqs = data.get('faqs', []) if isinstance(data, dict) else []
if faqs:
faq_summaries = [f"- Q: {faq.get('question', '')[:50]}... A: {faq.get('answer', '')[:50]}..." for faq in faqs[:3]]
summary = f"Found {len(faqs)} FAQ items:\n" + "\n".join(faq_summaries)
tool_context.append(summary)
else:
tool_context.append("No FAQ items found")
elif tool_name == "get_categories":
# Extract category names only
categories = data.get('categories', []) if isinstance(data, dict) else []
category_names = [cat.get('name', '') for cat in categories[:5] if cat.get('name')]
summary = f"Available categories: {', '.join(category_names)}"
if len(categories) > 5:
summary += f" (and {len(categories) - 5} more)"
tool_context.append(summary)
else:
# For other tools, include concise summary (limit to 200 chars)
data_str = json.dumps(data, ensure_ascii=False)[:200]
tool_context.append(f"Tool {tool_name} returned: {data_str}...")
else:
tool_context.append(f"Tool {result['tool_name']} failed: {result['error']}")
@@ -357,7 +440,8 @@ User question: {state["current_message"]}
Tool returned information:
{chr(10).join(tool_context)}
Please generate a friendly and professional response. If the tool did not return useful information, honestly inform the user and suggest other ways to get help.
Please generate a friendly and professional response in Chinese. Keep it concise but informative.
If the tool did not return useful information, honestly inform the user and suggest other ways to get help.
Return only the response content, do not return JSON."""
messages = [
@@ -367,11 +451,12 @@ Return only the response content, do not return JSON."""
try:
llm = get_llm_client()
response = await llm.chat(messages, temperature=0.7)
# Lower temperature for faster response
response = await llm.chat(messages, temperature=0.3)
state = set_response(state, response.content)
return state
except Exception as e:
logger.error("Response generation failed", error=str(e))
state = set_response(state, "Sorry, there was a problem processing your request. Please try again later or contact customer support.")
state = set_response(state, "抱歉，处理您的请求时出现问题。请稍后重试或联系人工客服。")
return state

197
agent/agents/product.py
View File

@@ -30,9 +30,11 @@ PRODUCT_AGENT_PROMPT = """你是一个专业的 B2B 商品顾问助手。
2. **get_product_detail** - 获取商品详情
- product_id: 商品ID
3. **recommend_products** - 智能推荐
- context: 推荐上下文（可包含当前查询、浏览历史等）
- limit: 推荐数量
3. **recommend_products** - 智能推荐（心动清单/猜你喜欢）
- page_size: 推荐数量（默认 6，最大 100）
- page: 页码（默认 1）
- warehouse_id: 仓库ID（默认 2）
- 说明：此工具使用 Mall API /mall/api/loveList 接口，需要用户 token 认证，系统会自动注入用户 token
4. **get_quote** - B2B 询价
- product_id: 商品ID
@@ -81,6 +83,18 @@ PRODUCT_AGENT_PROMPT = """你是一个专业的 B2B 商品顾问助手。
}
```
用户说："推荐一些商品"
返回：
```json
{
"action": "call_tool",
"tool_name": "recommend_products",
"arguments": {
"page_size": 6
}
}
```
当需要向用户询问更多信息时：
```json
{
@@ -104,6 +118,17 @@ PRODUCT_AGENT_PROMPT = """你是一个专业的 B2B 商品顾问助手。
- 报价通常有有效期
## 商品推荐策略
**重要规则：推荐 vs 搜索**
- **泛泛推荐**（"推荐一些商品"、"推荐一下"、"有什么好推荐的"） → 使用 recommend_products
- **具体商品推荐**（"推荐ring相关的商品"、"推荐手机"、"推荐一些珠宝"） → 使用 search_products (提取关键词：ring、手机、珠宝)
- **商品搜索**（"搜索ring"、"找ring商品"） → 使用 search_products
**说明**：
- 如果用户推荐请求中包含具体的商品关键词（如 ring、手机、珠宝等），使用 search_products 进行精准搜索
- 只有在泛泛请求推荐时才使用 recommend_products（基于用户行为的个性化推荐）
**其他推荐依据**：
- 根据用户采购历史推荐
- 根据当前查询语义推荐
- 根据企业行业特点推荐
@@ -119,12 +144,12 @@ PRODUCT_AGENT_PROMPT = """你是一个专业的 B2B 商品顾问助手。
async def product_agent(state: AgentState) -> AgentState:
"""Product agent node
Handles product search, recommendations, quotes and inventory queries.
Args:
state: Current agent state
Returns:
Updated state with tool calls or response
"""
@@ -133,11 +158,40 @@ async def product_agent(state: AgentState) -> AgentState:
conversation_id=state["conversation_id"],
sub_intent=state.get("sub_intent")
)
state["current_agent"] = "product"
state["agent_history"].append("product")
state["state"] = ConversationState.PROCESSING.value
# ========== FAST PATH: Image Search ==========
# Check if this is an image search request
image_search_url = state.get("context", {}).get("image_search_url")
if image_search_url:
logger.info(
"Image search detected, calling search_products_by_image",
conversation_id=state["conversation_id"],
image_url=image_search_url[:100] + "..." if len(image_search_url) > 100 else image_search_url
)
# 直接调用图片搜索工具
state = add_tool_call(
state,
tool_name="search_products_by_image",
arguments={
"image_url": image_search_url,
"page_size": 6,
"page": 1
},
server="product"
)
# 清除 image_search_url 防止无限循环
state["context"]["image_search_url"] = None
state["state"] = ConversationState.TOOL_CALLING.value
return state
# ==============================================
# Check if we have tool results to process
if state["tool_results"]:
return await _generate_product_response(state)
@@ -148,7 +202,8 @@ async def product_agent(state: AgentState) -> AgentState:
]
# Add conversation history
for msg in state["messages"][-6:]:
# 只保留最近 2 条历史消息以减少 token 数量和响应时间
for msg in state["messages"][-2:]:
messages.append(Message(role=msg["role"], content=msg["content"]))
# Build context info
@@ -233,7 +288,7 @@ async def product_agent(state: AgentState) -> AgentState:
# Set default page_size if not provided
if "page_size" not in arguments:
arguments["page_size"] = 5
arguments["page_size"] = 6
# Set default page if not provided
if "page" not in arguments:
@@ -249,8 +304,21 @@ async def product_agent(state: AgentState) -> AgentState:
# Inject context for recommendation
if tool_name == "recommend_products":
arguments["user_id"] = state["user_id"]
arguments["account_id"] = state["account_id"]
arguments["user_token"] = state.get("user_token")
# 如果没有提供 page_size，使用默认值 6
if "page_size" not in arguments:
arguments["page_size"] = 6
# 如果没有提供 warehouse_id，使用默认值 2
if "warehouse_id" not in arguments:
arguments["warehouse_id"] = 2
logger.info(
"Product agent recommend_products after injection",
user_token_present="user_token" in arguments,
user_token_preview=arguments.get("user_token", "")[:20] + "..." if arguments.get("user_token") else None,
arguments=arguments,
conversation_id=state["conversation_id"]
)
# Inject context for quote
if tool_name == "get_quote":
@@ -301,25 +369,55 @@ async def product_agent(state: AgentState) -> AgentState:
async def _generate_product_response(state: AgentState) -> AgentState:
"""Generate response based on product tool results"""
# 特殊处理：如果是 search_products 工具返回，直接发送商品卡片
has_product_search_result = False
# 特殊处理：如果是 search_products、recommend_products 或 search_products_by_image 工具返回，直接发送商品卡片
has_product_result = False
products = []
result_source = None # "search", "recommend" 或 "image_search"
# 添加日志：查看所有工具结果
import json as json_module
logger.info(
"All tool results",
tool_results_count=len(state.get("tool_results", [])),
tool_results=json_module.dumps(state.get("tool_results", []), ensure_ascii=False, indent=2)[:2000]
)
for result in state["tool_results"]:
if result["success"] and result["tool_name"] == "search_products":
logger.info(
"Processing tool result",
tool_name=result["tool_name"],
success=result["success"],
data_keys=list(result.get("data", {}).keys()) if isinstance(result.get("data"), dict) else "not a dict",
data_preview=json_module.dumps(result.get("data"), ensure_ascii=False)[:500]
)
if result["success"] and result["tool_name"] in ["search_products", "recommend_products", "search_products_by_image"]:
data = result["data"]
if isinstance(data, dict) and data.get("success"):
products = data.get("products", [])
has_product_search_result = True
# MCP 返回的数据结构: {"success": true, "result": {"success": true, "products": [...]}}
# 需要从 result.result 中提取实际数据
inner_data = data.get("result", data)
products = inner_data.get("products", [])
keyword = inner_data.get("keyword", "")
has_product_result = True
if result["tool_name"] == "recommend_products":
result_source = "recommend"
elif result["tool_name"] == "search_products_by_image":
result_source = "image_search"
else:
result_source = "search"
logger.info(
"Product search results found",
f"Product {result_source} results found",
products_count=len(products),
keyword=data.get("keyword", "")
keyword=keyword,
products_preview=json_module.dumps(products[:2], ensure_ascii=False, indent=2) if products else "[]"
)
break
# 如果有商品搜索结果，直接发送商品卡片
if has_product_search_result and products:
# 如果有商品结果，直接发送商品卡片（product_list 格式）
if has_product_result and products:
try:
from integrations.chatwoot import ChatwootClient
from core.language_detector import detect_language
@@ -327,7 +425,7 @@ async def _generate_product_response(state: AgentState) -> AgentState:
# 检测语言
detected_language = state.get("detected_language", "en")
# 发送商品卡片
# 发送商品列表
chatwoot = ChatwootClient(account_id=int(state.get("account_id", 1)))
conversation_id = state.get("conversation_id")
@@ -339,10 +437,11 @@ async def _generate_product_response(state: AgentState) -> AgentState:
)
logger.info(
"Product cards sent successfully",
f"Product {result_source} cards sent successfully",
conversation_id=conversation_id,
products_count=len(products),
language=detected_language
language=detected_language,
result_source=result_source
)
# 清空响应，避免重复发送
@@ -352,17 +451,56 @@ async def _generate_product_response(state: AgentState) -> AgentState:
except Exception as e:
logger.error(
"Failed to send product cards, falling back to text response",
f"Failed to send product {result_source} cards, falling back to text response",
error=str(e),
products_count=len(products)
products_count=len(products),
result_source=result_source
)
# 常规处理：生成文本响应
tool_context = []
for result in state["tool_results"]:
if result["success"]:
data = result["data"]
tool_context.append(f"工具 {result['tool_name']} 返回:\n{json.dumps(data, ensure_ascii=False, indent=2)}")
tool_name = result['tool_name']
data = result['data']
# Extract only essential information based on tool type
if tool_name == "search_products" or tool_name == "recommend_products":
products = data.get("products", []) if isinstance(data, dict) else []
if products:
product_summaries = [f"- {p.get('product_name', 'N/A')}: {p.get('price', 'N/A')}" for p in products[:3]]
summary = f"Found {len(products)} products:\n" + "\n".join(product_summaries)
if len(products) > 3:
summary += f"\n(and {len(products) - 3} more)"
tool_context.append(summary)
else:
tool_context.append("No products found")
elif tool_name == "get_product_detail":
product = data.get("product", {}) if isinstance(data, dict) else {}
name = product.get("product_name", product.get("name", "N/A"))
price = product.get("price", "N/A")
stock = product.get("stock", product.get("stock_status", "N/A"))
summary = f"Product: {name} | Price: {price} | Stock: {stock}"
tool_context.append(summary)
elif tool_name == "check_inventory":
inventory = data.get("inventory", []) if isinstance(data, dict) else []
inv_summaries = [f"{inv.get('product_id', 'N/A')}: {inv.get('quantity', 'N/A')} available" for inv in inventory[:3]]
summary = "Inventory status:\n" + "\n".join(inv_summaries)
tool_context.append(summary)
elif tool_name == "get_pricing":
product_id = data.get("product_id", "N/A")
unit_price = data.get("unit_price", "N/A")
total_price = data.get("total_price", "N/A")
summary = f"Quote for {product_id}: Unit: {unit_price} | Total: {total_price}"
tool_context.append(summary)
else:
# For other tools, include concise summary (limit to 200 chars)
data_str = json.dumps(data, ensure_ascii=False)[:200]
tool_context.append(f"工具 {tool_name} 返回: {data_str}...")
# Extract product context
if isinstance(data, dict):
@@ -398,7 +536,8 @@ async def _generate_product_response(state: AgentState) -> AgentState:
try:
llm = get_llm_client()
response = await llm.chat(messages, temperature=0.7)
# Lower temperature for faster response
response = await llm.chat(messages, temperature=0.3)
state = set_response(state, response.content)
return state