Update app.py

app.py

@@ -3,7 +3,6 @@ import requests
 import os
 import fitz
 import re
-import time
 from huggingface_hub import HfApi, hf_hub_download
 
 # --- 核心配置 ---
@@ -11,16 +10,18 @@ GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")
 HF_TOKEN = os.environ.get("HF_TOKEN")
 MODEL_ID = "gemini-3-flash-preview"
 
-# 包含您新加的 TaxGuides
 DATASETS = ["MGGroup/Treaties", "MGGroup/InvestmentGuide"]
 
+# 顶部的专业描述（保持原样，符合您的要求）
 DESCRIPTION = """
 <div style="text-align: left; border-left: 4px solid #2196F3; padding-left: 15px; margin-bottom: 20px;">
     <h3>MG TaxAI | 跨境财税合规实验室 (Beta)</h3>
-    <p>本系统依托 <b>MG 核心智库</b> 构建，已深度整合 2025/2026 全球税务指南。支持解析结果实时溯源至官方文本。</p>
+    <p>本系统依托 <b>MG 核心智库</b> 构建，旨在实现解析结果实时溯源至各国官方税收协定与法律文本。目前系统正处于<b>知识库全量装载阶段</b>，已优先上线核心业务国家的官方协定库。</p>
+    <p>我们正持续同步全球各主要经济体的国别投资税收指南及多税种年度税收报告。受限于测试版的数据填充进度，相关解析结果仅供专业参考。MG 团队正加速完善每一条咨询建议的合规证据链，以确保交付专家级的数字化合规支持。</p>
     <hr style="border: 0; border-top: 1px solid #eee; margin: 10px 0;">
     <p style="font-size: 0.85em; color: #666;">
-        <b>⚠️ AI 免责声明：</b> 仅供专业参考，决策前请咨询 MG Consult 专业团队。
+        <b>⚠️ AI 免责声明：</b><br>
+        本系统生成的内容由人工智能根据现有库文件分析得出，不构成正式的法律或税务建议。在使用本系统结果进行任何商业决策前，请务必咨询 MG Consult 专业团队。
     </p>
 </div>
 """
@@ -29,79 +30,46 @@ def fetch_dataset_context(query):
     if not HF_TOKEN: return ""
     api = HfApi(token=HF_TOKEN)
     combined_text = ""
-    
-    # 提取国家和业务关键词
     keywords = [k for k in re.findall(r'[\u4e00-\u9fa5]{2,}|[a-zA-Z]{3,}', query) if k not in ["资料", "关于", "查询", "政策"]]
-    # 针对 Worldwide 指南的主题映射
-    topic_keywords = []
-    if "个人" in query or "人所得" in query: topic_keywords.append("Personal")
-    if "所得税" in query: topic_keywords.append("Corporate")
-    if "定价" in query: topic_keywords.append("Transfer")
-
+    if not keywords: keywords = re.findall(r'[\u4e00-\u9fa5]+|[a-zA-Z]+', query)
+    
     for repo in DATASETS:
         try:
             files = api.list_repo_files(repo_id=repo, repo_type="dataset")
-            # 匹配文件名：包含国家关键词 或 包含主题词
-            matched = sorted([f for f in files if f.lower().endswith(".pdf") and any(k.lower() in f.lower() for k in keywords + topic_keywords)])
-            
-            for f_path in matched[:5]: # 减少到5个文件以降低负荷
-                try:
-                    temp_path = hf_hub_download(repo_id=repo, filename=f_path, repo_type="dataset", token=HF_TOKEN)
-                    doc = fitz.open(temp_path)
-                    
-                    # --- 核心优化：页码精确定位 ---
-                    pages_to_read = []
-                    country_key = next((k for k in keywords if len(k) >= 2), "")
-                    
-                    if country_key: # 如果提到了国家，只读含国家名的页
-                        for i in range(len(doc)):
-                            if country_key in doc[i].get_text():
-                                pages_to_read.append(i)
-                            if len(pages_to_read) >= 10: break
-                    
-                    if not pages_to_read: pages_to_read = range(min(15, len(doc)))
-                    
-                    combined_text += f"\n[参考文件: {f_path}]\n"
-                    for p in pages_to_read:
-                        combined_text += doc[p].get_text()
-                    doc.close()
-                    os.remove(temp_path)
-                except: continue
+            matched = sorted([f for f in files if f.lower().endswith(".pdf") and any(k.lower() in f.lower() for k in keywords)])
+            for f_path in matched[:8]: 
+                temp_path = hf_hub_download(repo_id=repo, filename=f_path, repo_type="dataset", token=HF_TOKEN)
+                doc = fitz.open(temp_path)
+                combined_text += f"\n[Ref: {f_path}]\n" + "".join([page.get_text() for page in doc[:15]])
+                doc.close()
+                os.remove(temp_path)
         except: continue
-    return combined_text[:8000] # 压缩上下文到 8000，显著提升 API 成功率
+    return combined_text[:12000]
 
-def call_gemini_with_retry(messages):
-    url = f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL_ID}:generateContent?key={GEMINI_API_KEY}"
-    payload = {
-        "contents": [{"role": "user" if m["role"] in ["user", "system"] else "model", 
-                     "parts": [{"text": m["content"]}]} for m in messages],
-        "tools": [{"google_search_retrieval": {}}]
-    }
-    
-    for attempt in range(3):
-        try:
-            res = requests.post(url, json=payload, timeout=90)
-            if res.status_code == 200:
-                return res.json()['candidates'][0]['content']['parts'][0]['text']
-            elif res.status_code == 429:
-                time.sleep(10 * (attempt + 1)) # 遭遇 429 自动阶梯式等待
-                continue
-            else:
-                return f"⚠️ 引擎响应异常 (Code: {res.status_code})"
-        except:
-            time.sleep(5)
-            continue
-    return "⚠️ 系统目前过于繁忙，请 30 秒后重试。"
+def find_local_context(query):
+    base_dir = "./treaties"
+    if not os.path.exists(base_dir): return ""
+    try:
+        keywords = re.findall(r'[\u4e00-\u9fa5]+|[a-zA-Z]+', query)
+        folders = [d for d in os.listdir(base_dir) if os.path.isdir(os.path.join(base_dir, d))]
+        selected = next((f for f in folders if any(k.lower() in f.lower() for k in keywords)), None)
+        if not selected: return ""
+        path = os.path.join(base_dir, selected)
+        texts = [f"\n[Local: {pdf}]\n" + "".join([p.get_text() for p in fitz.open(os.path.join(path, pdf))[:15]]) for pdf in sorted([f for f in os.listdir(path) if f.endswith(".pdf")])[:5]]
+        return "\n".join(texts)[:8000]
+    except: return ""
 
 def respond(message, history, system_message, max_tokens, temperature, top_p):
-    # 获取压缩后的知识
     knowledge = fetch_dataset_context(message)
+    if not knowledge: knowledge = find_local_context(message)
     
+    # --- 核心修改：在系统提示词中强制要求 AI 保持静默，不要自我介绍 ---
     full_system_prompt = (
         f"{system_message}\n\n"
         "【重要约束】：\n"
-        "1. 智库包含 Worldwide 指南，请准确定位该国家章节并结合 Google Search 验证 2026 最新政策。\n"
-        "2. 严禁自我介绍，直接进入分析。\n\n"
+        "1. 严禁进行任何形式的自我介绍或身份说明（例如：不要说'我是MG的专家'、'你好'等）。\n"
+        "2. 严禁包含任何开场白，直接针对用户问题进入专业分析。\n"
+        "3. 必须严格基于以下参考资料进行回答。\n\n"
         f"参考资料：\n{knowledge}"
     )
 
@@ -111,21 +79,28 @@ def respond(message, history, system_message, max_tokens, temperature, top_p):
         if a: messages.append({"role": "assistant", "content": a})
     messages.append({"role": "user", "content": message})
 
-    reply = call_gemini_with_retry(messages)
+    url = f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL_ID}:generateContent?key={GEMINI_API_KEY}"
+    try:
+        res = requests.post(url, json={"contents": [{"role":"user" if m["role"] in ["user","system"] else "model", "parts":[{"text":m["content"]}]} for m in messages]}, timeout=60)
+        reply = res.json()['candidates'][0]['content']['parts'][0]['text']
+    except:
+        reply = "系统繁忙，请稍后再试。"
 
+    # --- 唯一的身份开场白在这里定义，由代码控制，不给 AI 发挥空间 ---
     header = "您好，我是 **MG Consult** 的国际税收 AI 专家。很高兴为您提供专业咨询。\n\n---\n\n"
     yield header + reply
 
+# 界面
 demo = gr.ChatInterface(
     fn=respond,
     description=DESCRIPTION,
     theme="soft",
-    css=".gradio-container {max-width: 950px !important}",
+    css=".gradio-container {max-width: 950px !important} .description {margin-bottom: 20px}",
     additional_inputs=[
-        gr.Textbox(value="你代表 MG Consult 专家。直接进入正文，严禁自我介绍。", label="系统指令"),
-        gr.Slider(512, 4096, 2048, label="回复长度"),
-        gr.Slider(0, 1, 0.05, label="严谨度"),
-        gr.Slider(0, 1, 0.95, label="采样率"),
+        gr.Textbox(value="你代表 MG Consult，是国际税收专家。请严格基于参考资料提供深度分析。请直接进入正文，严禁自我介绍。", label="系统指令"),
+        gr.Slider(512, 4096, 2048, label="回复长度限制"),
+        gr.Slider(0, 1, 0.05, label="严谨度 (Temperature)"),
+        gr.Slider(0, 1, 0.95, label="采样率 (Top-p)"),
     ],
 )