search_tools.py
5.94 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
"""
Search Tools for Product Discovery
Provides text-based search via Search API and VLM style analysis
"""
import base64
import logging
from pathlib import Path
from typing import Optional
import requests
from langchain_core.tools import tool
from openai import OpenAI
from app.config import settings
logger = logging.getLogger(__name__)
_openai_client: Optional[OpenAI] = None
def get_openai_client() -> OpenAI:
global _openai_client
if _openai_client is None:
kwargs = {"api_key": settings.openai_api_key}
if settings.openai_api_base_url:
kwargs["base_url"] = settings.openai_api_base_url
_openai_client = OpenAI(**kwargs)
return _openai_client
@tool
def search_products(query: str, limit: int = 5) -> str:
"""Search for fashion products using natural language descriptions.
Use when users describe what they want:
- "Find me red summer dresses"
- "Show me blue running shoes"
- "I want casual shirts for men"
Args:
query: Natural language product description
limit: Maximum number of results (1-20)
Returns:
Formatted string with product information
"""
try:
logger.info(f"Searching products: '{query}', limit: {limit}")
url = f"{settings.search_api_base_url.rstrip('/')}/search/"
headers = {
"Content-Type": "application/json",
"X-Tenant-ID": settings.search_api_tenant_id,
}
payload = {
"query": query,
"size": min(limit, 20),
"from": 0,
"language": "zh",
}
response = requests.post(url, json=payload, headers=headers, timeout=60)
if response.status_code != 200:
logger.error(f"Search API error: {response.status_code} - {response.text}")
return f"Error searching products: API returned {response.status_code}"
data = response.json()
results = data.get("results", [])
if not results:
return "No products found matching your search."
output = f"Found {len(results)} product(s):\n\n"
for idx, product in enumerate(results, 1):
output += f"{idx}. {product.get('title', 'Unknown Product')}\n"
output += f" ID: {product.get('spu_id', 'N/A')}\n"
output += f" Category: {product.get('category_path', product.get('category_name', 'N/A'))}\n"
if product.get("vendor"):
output += f" Brand: {product.get('vendor')}\n"
if product.get("price") is not None:
output += f" Price: {product.get('price')}\n"
# 规格/颜色信息
specs = product.get("specifications", [])
if specs:
color_spec = next(
(s for s in specs if s.get("name") == "color"),
None,
)
if color_spec:
output += f" Color: {color_spec.get('value', 'N/A')}\n"
if product.get("relevance_score") is not None:
output += f" Relevance: {product['relevance_score']:.2f}\n"
output += "\n"
return output.strip()
except requests.exceptions.RequestException as e:
logger.error(f"Error searching products (network): {e}", exc_info=True)
return f"Error searching products: {str(e)}"
except Exception as e:
logger.error(f"Error searching products: {e}", exc_info=True)
return f"Error searching products: {str(e)}"
@tool
def analyze_image_style(image_path: str) -> str:
"""Analyze a fashion product image using AI vision to extract detailed style information.
Use when you need to understand style/attributes from an image:
- Understand the style, color, pattern of a product
- Extract attributes like "casual", "formal", "vintage"
- Get detailed descriptions for subsequent searches
Args:
image_path: Path to the image file
Returns:
Detailed text description of the product's visual attributes
"""
try:
logger.info(f"Analyzing image with VLM: '{image_path}'")
img_path = Path(image_path)
if not img_path.exists():
return f"Error: Image file not found at '{image_path}'"
with open(img_path, "rb") as image_file:
image_data = base64.b64encode(image_file.read()).decode("utf-8")
prompt = """Analyze this fashion product image and provide a detailed description.
Include:
- Product type (e.g., shirt, dress, shoes, pants, bag)
- Primary colors
- Style/design (e.g., casual, formal, sporty, vintage, modern)
- Pattern or texture (e.g., plain, striped, checked, floral)
- Key features (e.g., collar type, sleeve length, fit)
- Material appearance (if obvious, e.g., denim, cotton, leather)
- Suitable occasion (e.g., office wear, party, casual, sports)
Provide a comprehensive yet concise description (3-4 sentences)."""
client = get_openai_client()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}",
"detail": "high",
},
},
],
}
],
max_tokens=500,
temperature=0.3,
)
analysis = response.choices[0].message.content.strip()
logger.info("VLM analysis completed")
return analysis
except Exception as e:
logger.error(f"Error analyzing image: {e}", exc_info=True)
return f"Error analyzing image: {str(e)}"
def get_all_tools():
"""Get all available tools for the agent"""
return [search_products, analyze_image_style]