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 底部导航修复记录 - 新增历史变更条目
61 lines
3.0 KiB
Python
61 lines
3.0 KiB
Python
from typing import Dict, Any, Optional
|
|
import json, httpx
|
|
from app.ai.base import AIProvider
|
|
|
|
|
|
class LocalProvider(AIProvider):
|
|
def __init__(self, model_url: str = "http://localhost:8001", model_name: str = "gemma-3-8b"):
|
|
self.model_url = model_url.rstrip("/")
|
|
self.model_name = model_name
|
|
self._name = f"local-{model_name}"
|
|
|
|
async def translate(self, text: str, source_lang: Optional[str], target_lang: str, context: Optional[str] = None) -> Dict[str, Any]:
|
|
prompt = f"Translate{ f' from {source_lang}' if source_lang else ''} to {target_lang}:\n{text}\n\nTranslation:"
|
|
result = await self._generate(prompt)
|
|
return {"translated_text": result, "provider": self.name, "cost": 0.0}
|
|
|
|
async def reply(self, inquiry: str, context: Optional[Dict[str, Any]] = None, tone: str = "professional", preference_context: Optional[str] = None) -> Dict[str, Any]:
|
|
prompt = ""
|
|
if preference_context:
|
|
prompt += f"[User prefers: {preference_context}]\n"
|
|
if context:
|
|
prompt += "\n".join(f"{k}: {v}" for k, v in context.items() if v) + "\n"
|
|
prompt += f"Customer: {inquiry}\n\nWrite a {tone} reply:"
|
|
result = await self._generate(prompt)
|
|
return {"reply": result, "provider": self.name, "cost": 0.0}
|
|
|
|
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]:
|
|
info = json.dumps(product_info, ensure_ascii=False)
|
|
prompt = ""
|
|
if preference_context:
|
|
prompt += f"[User prefers: {preference_context}]\n"
|
|
prompt += f"Product: {info}\nTarget: {target}\nStyle: {style}\nLanguage: {language}\n\nMarketing copy:"
|
|
result = await self._generate(prompt, max_tokens=800)
|
|
return {"content": result, "provider": self.name, "cost": 0.0}
|
|
|
|
async def extract_info(self, text: str, schema: Dict[str, Any]) -> Dict[str, Any]:
|
|
prompt = f"Extract JSON from text matching schema:\nSchema: {json.dumps(schema)}\n\nText: {text}\n\nJSON:"
|
|
result = await self._generate(prompt, max_tokens=500)
|
|
try:
|
|
return {"data": json.loads(result), "confidence": 0.7, "provider": self.name, "cost": 0.0}
|
|
except json.JSONDecodeError:
|
|
return {"data": {}, "confidence": 0.0, "provider": self.name, "cost": 0.0, "error": "parse_failed"}
|
|
|
|
async def _generate(self, prompt: str, max_tokens: int = 500) -> str:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{self.model_url}/v1/completions",
|
|
json={"model": self.model_name, "prompt": prompt, "max_tokens": max_tokens, "temperature": 0.7, "stream": False},
|
|
timeout=60,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()["choices"][0]["text"].strip()
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def cost_per_1k_tokens(self) -> float:
|
|
return 0.0
|