| import os |
| import base64 |
| import gradio as gr |
| from openai import OpenAI |
|
|
| def encode_image(image_path): |
| """将本地图片文件转换为 Base64 编码""" |
| with open(image_path, "rb") as image_file: |
| return base64.b64encode(image_file.read()).decode('utf-8') |
|
|
| def detect_language(text, history): |
| """ |
| 智能检测用户语言： |
| 1. 优先从当前提问检测 |
| 2. 若当前无文字，从最近一轮用户历史中推断 |
| 3. 默认返回中文 |
| """ |
| |
| if text: |
| clean_text = str(text).strip() |
| if clean_text: |
| |
| chinese_chars = sum(1 for char in clean_text if '\u4e00' <= char <= '\u9fff') |
| alpha_chars = sum(1 for char in clean_text if char.isalpha()) |
| |
| if alpha_chars > 0: |
| |
| if chinese_chars / alpha_chars > 0.3: |
| return "zh" |
| else: |
| return "en" |
| else: |
| |
| return "zh" |
| |
| |
| for msg in reversed(history[-6:]): |
| if msg.get("role") == "user": |
| content = msg.get("content", "") |
| if isinstance(content, dict): |
| content = content.get("text", "") |
| if content: |
| clean_content = str(content).strip() |
| chinese_chars = sum(1 for char in clean_content if '\u4e00' <= char <= '\u9fff') |
| alpha_chars = sum(1 for char in clean_content if char.isalpha()) |
| if alpha_chars > 0 and chinese_chars / alpha_chars > 0.3: |
| return "zh" |
| elif alpha_chars > 0: |
| return "en" |
| |
| |
| return "zh" |
|
|
| def predict(message, history): |
| api_key = os.getenv("SILICONFLOW_API_KEY", "") |
| if not api_key: |
| yield "❌ 未检测到 API Key！请先在 Space Settings -> Secrets 中添加 SILICONFLOW_API_KEY！" |
| return |
|
|
| api_key = api_key.strip() |
|
|
| try: |
| client = OpenAI( |
| api_key=api_key, |
| base_url="https://api.siliconflow.cn/v1" |
| ) |
|
|
| text_content = "" |
| files = [] |
|
|
| |
| if isinstance(message, dict): |
| text_content = message.get("text", "") |
| files = message.get("files", []) |
| else: |
| text_content = str(message) |
|
|
| |
| user_lang = detect_language(text_content, history) |
| |
| |
| print(f"检测到语言: {user_lang}, 用户输入: {text_content[:50]}") |
|
|
| |
| current_content = [] |
|
|
| |
| for f in files: |
| file_path = f.get("path") if isinstance(f, dict) else f |
| if file_path: |
| try: |
| base64_img = encode_image(file_path) |
| current_content.append({ |
| "type": "image_url", |
| "image_url": {"url": f"data:image/jpeg;base64,{base64_img}"} |
| }) |
| except Exception as img_err: |
| print(f"图片读取失败: {img_err}") |
|
|
| |
| if not text_content or not str(text_content).strip(): |
| if user_lang == "en": |
| prompt_text = "Please observe this image carefully and tell me what plant this is and its key features." |
| else: |
| prompt_text = "请仔细观察这张图片，告诉我这是什么植物？有哪些特征？" |
| else: |
| prompt_text = str(text_content).strip() |
|
|
| current_content.append({"type": "text", "text": prompt_text}) |
|
|
| |
| if user_lang == "en": |
| system_prompt = ( |
| "You are a professional botanist and nature expert. " |
| "Identify the plant in the image and provide its scientific name, " |
| "family, genus, and key features in **English**. " |
| "Be concise and accurate." |
| ) |
| else: |
| system_prompt = ( |
| "你是一位专业的植物学家和自然科普专家。" |
| "请精准识别图片中的植物，并用简洁明了的中文回答其名称、科属及特点。" |
| ) |
|
|
| messages = [ |
| {"role": "system", "content": system_prompt} |
| ] |
|
|
| |
| for msg in history[-4:]: |
| role = "user" if msg.get("role") == "user" else "assistant" |
| content_val = msg.get("content", "") |
| if isinstance(content_val, dict): |
| content_val = content_val.get("text", "") |
| if content_val and not str(content_val).startswith("data:image"): |
| messages.append({"role": role, "content": str(content_val)}) |
|
|
| |
| messages.append({"role": "user", "content": current_content}) |
|
|
| |
| response = client.chat.completions.create( |
| model="Qwen/Qwen3-VL-32B-Instruct", |
| messages=messages, |
| stream=True, |
| temperature=0.2 |
| ) |
|
|
| |
| partial_text = "" |
| for chunk in response: |
| if chunk.choices and chunk.choices[0].delta.content: |
| partial_text += chunk.choices[0].delta.content |
| yield partial_text |
|
|
| except Exception as e: |
| err_msg = str(e) |
| if "30001" in err_msg or "insufficient" in err_msg.lower(): |
| yield "⚠️ **请求失败：硅基流动账号余额不足**" |
| elif "30003" in err_msg or "Model disabled" in err_msg: |
| yield "⚠️ **模型不可用，请检查模型名称是否正确**" |
| else: |
| yield f"❌ 请求失败: {err_msg}" |
|
|
| |
| demo = gr.ChatInterface( |
| fn=predict, |
| title="🌱 AI 植物识别专家 (智能双语)", |
| description="🌍 智能识别你的提问语言：中文问→中文答，英文问→英文答。连续对话自动记忆语言偏好！", |
| multimodal=True |
| ) |
|
|
| if __name__ == "__main__": |
| demo.launch(ssr_mode=False) |
