Blame view

app/tools/search_tools.py 5.94 KB
e7f2b240   tangwang   first commit
1
2
  """
  Search Tools for Product Discovery
8810a6fa   tangwang   重构
3
  Provides text-based search via Search API and VLM style analysis
e7f2b240   tangwang   first commit
4
5
6
7
8
9
10
  """
  
  import base64
  import logging
  from pathlib import Path
  from typing import Optional
  
8810a6fa   tangwang   重构
11
  import requests
e7f2b240   tangwang   first commit
12
13
14
15
  from langchain_core.tools import tool
  from openai import OpenAI
  
  from app.config import settings
e7f2b240   tangwang   first commit
16
17
18
  
  logger = logging.getLogger(__name__)
  
e7f2b240   tangwang   first commit
19
20
21
  _openai_client: Optional[OpenAI] = None
  
  
e7f2b240   tangwang   first commit
22
23
24
  def get_openai_client() -> OpenAI:
      global _openai_client
      if _openai_client is None:
8810a6fa   tangwang   重构
25
26
27
28
          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)
e7f2b240   tangwang   first commit
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
      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}")
  
8810a6fa   tangwang   重构
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
          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", [])
e7f2b240   tangwang   first commit
71
72
73
74
75
76
77
  
          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):
8810a6fa   tangwang   重构
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
              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"
e7f2b240   tangwang   first commit
98
99
100
101
102
  
              output += "\n"
  
          return output.strip()
  
8810a6fa   tangwang   重构
103
104
105
      except requests.exceptions.RequestException as e:
          logger.error(f"Error searching products (network): {e}", exc_info=True)
          return f"Error searching products: {str(e)}"
e7f2b240   tangwang   first commit
106
107
108
109
110
111
      except Exception as e:
          logger.error(f"Error searching products: {e}", exc_info=True)
          return f"Error searching products: {str(e)}"
  
  
  @tool
e7f2b240   tangwang   first commit
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
  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"""
8810a6fa   tangwang   重构
183
      return [search_products, analyze_image_style]