TradeMate Dev
7b62c2f8b4
## H5 底部导航修复 (Bug #10) - 精简 App.vue，移除重复 tabbar，仅保留全局样式 - uni-page 设置 height: calc(100% - 50px) + overflow-y: auto - 内容区域精确停在底部导航上方，独立滚动不再叠加 - 恢复 custom-tab-bar 组件 ## 项目进度文档 - PROGRESS.md 更新至 10 个 Bug 修复 - 新增 H5 底部导航修复记录 - 新增历史变更条目
88 lines
4.0 KiB
Python
88 lines
4.0 KiB
Python
from typing import Dict, Any, Optional
|
|
import json
|
|
from openai import AsyncOpenAI
|
|
from app.ai.base import AIProvider
|
|
|
|
|
|
SYSTEM_PROMPTS = {
|
|
"translate": "You are a professional translator specialized in foreign trade. "
|
|
"Translate business terms accurately. Return ONLY the translated text.",
|
|
"reply": "You are an experienced foreign trade sales expert. Write professional, "
|
|
"clear business replies. Return ONLY the reply text.",
|
|
"marketing": "You are a creative copywriter for international trade. "
|
|
"Return ONLY the marketing copy, no explanations.",
|
|
"extract": "Extract structured data from text. Return ONLY valid JSON.",
|
|
}
|
|
|
|
|
|
class SparkProvider(AIProvider):
|
|
def __init__(self, api_key: str, model: str = "astron-code-latest", base_url: str = None):
|
|
from app.config import settings
|
|
self.client = AsyncOpenAI(
|
|
api_key=api_key,
|
|
base_url=base_url or settings.IFLYTEK_API_BASE,
|
|
)
|
|
self.model = model
|
|
self._name = f"spark-{model}"
|
|
|
|
async def translate(self, text: str, source_lang: Optional[str], target_lang: str, context: Optional[str] = None) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["translate"]
|
|
if context:
|
|
system += f"\nContext: {context}"
|
|
prompt = f"Translate {f'from {source_lang} ' if source_lang and source_lang != 'auto' else ''}to {target_lang}:\n\n{text}"
|
|
content = await self._call(system, prompt)
|
|
return {"translated_text": content, "provider": self.name}
|
|
|
|
async def reply(self, inquiry: str, context: Optional[Dict[str, Any]] = None, tone: str = "professional", preference_context: Optional[str] = None) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["reply"] + f"\nTone: {tone}"
|
|
if preference_context:
|
|
system += f"\nUser preference: {preference_context}"
|
|
ctx = ""
|
|
if context:
|
|
ctx = "\n".join(f"{k}: {v}" for k, v in context.items() if v)
|
|
prompt = f"{ctx}\nCustomer inquiry:\n{inquiry}\n\nWrite a reply:"
|
|
content = await self._call(system, prompt)
|
|
return {"reply": content, "provider": self.name}
|
|
|
|
async def generate_marketing(self, product_info: Dict[str, Any], target: str, style: str = "professional", language: str = "en", preference_context: Optional[str] = None) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["marketing"] + f"\nStyle: {style}\nAudience: {target}\nLanguage: {language}"
|
|
if preference_context:
|
|
system += f"\nUser preference: {preference_context}"
|
|
info = json.dumps(product_info, ensure_ascii=False)
|
|
prompt = f"Product:\n{info}\n\nGenerate marketing copy:"
|
|
content = await self._call(system, prompt, max_tokens=1500)
|
|
return {"content": content, "provider": self.name}
|
|
|
|
async def extract_info(self, text: str, schema: Dict[str, Any]) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["extract"]
|
|
prompt = f"Schema:\n{json.dumps(schema, indent=2)}\n\nText:\n{text}\n\nJSON:"
|
|
content = await self._call(system, prompt, response_format={"type": "json_object"})
|
|
try:
|
|
data = json.loads(content)
|
|
return {"data": data, "confidence": 0.9, "provider": self.name}
|
|
except json.JSONDecodeError:
|
|
return {"data": {}, "confidence": 0.0, "provider": self.name}
|
|
|
|
async def _call(self, system: str, prompt: str, max_tokens: int = 1000, response_format: Optional[Dict] = None) -> str:
|
|
kwargs = {
|
|
"model": self.model,
|
|
"messages": [
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"max_tokens": max_tokens,
|
|
"temperature": 0.7,
|
|
}
|
|
if response_format:
|
|
kwargs["response_format"] = response_format
|
|
resp = await self.client.chat.completions.create(**kwargs)
|
|
return resp.choices[0].message.content
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def cost_per_1k_tokens(self) -> float:
|
|
return 0.0
|