import gradio as gr import requests import json import os from huggingface_hub import InferenceClient # ========== 配置两个 MCP 服务 ========== MCP_SERVICES = { "financial": { "name": "SEC Financial Reports", "url": "https://jc321-easyreportdatamcp.hf.space/mcp", "tools": [ { "type": "function", "function": { "name": "advanced_search_company", "description": "Search for US listed companies by name or ticker to get CIK and basic info", "parameters": { "type": "object", "properties": { "company_input": {"type": "string", "description": "Company name or ticker (e.g., 'Apple', 'AAPL')"} }, "required": ["company_input"] } } }, { "type": "function", "function": { "name": "get_latest_financial_data", "description": "Get latest financial data (revenue, net income, EPS, etc.) for a company", "parameters": { "type": "object", "properties": { "cik": {"type": "string", "description": "10-digit CIK number"} }, "required": ["cik"] } } }, { "type": "function", "function": { "name": "extract_financial_metrics", "description": "Get multi-year financial trends (3 or 5 years)", "parameters": { "type": "object", "properties": { "cik": {"type": "string", "description": "10-digit CIK number"}, "years": {"type": "integer", "enum": [3, 5]} }, "required": ["cik", "years"] } } } ] }, "market": { "name": "Market & Stock Data", "url": "https://jc321-marketandstockmcp.hf.space/mcp", "tools": [] # 需要获取实际工具列表 } } # 合并所有工具 ALL_TOOLS = [] TOOL_ROUTING = {} # tool_name -> mcp_url 的映射 for service_key, service in MCP_SERVICES.items(): for tool in service["tools"]: tool_name = tool["function"]["name"] ALL_TOOLS.append(tool) TOOL_ROUTING[tool_name] = service["url"] # ========== 初始化 LLM 客户端 ========== hf_token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") client = InferenceClient(api_key=hf_token) if hf_token else InferenceClient() print(f"✅ LLM client initialized (Qwen/Qwen2.5-72B-Instruct:novita)") # ========== 系统提示词 ========== SYSTEM_PROMPT = """You are an intelligent financial and market analysis assistant with access to real-time data. You have access to two powerful data sources: 1. **SEC Financial Reports** - Get official financial data for US-listed companies (revenue, earnings, cash flow, etc.) 2. **Market & Stock Data** - Get real-time market data, stock prices, and market analysis When users ask about: - Company financials, earnings, revenue → Use SEC Financial Reports tools - Stock prices, market trends, trading data → Use Market & Stock Data tools Be conversational, insightful, and provide data-driven analysis. Automatically fetch data when needed.""" # ========== 核心函数：调用 MCP 工具 ========== def call_mcp_tool(tool_name, arguments): """调用 MCP 工具""" mcp_url = TOOL_ROUTING.get(tool_name) if not mcp_url: return {"error": f"Unknown tool: {tool_name}"} try: response = requests.post( mcp_url, json={ "jsonrpc": "2.0", "method": "tools/call", "params": {"name": tool_name, "arguments": arguments}, "id": 1 }, headers={"Content-Type": "application/json"}, timeout=60 ) if response.status_code == 200: return response.json() else: return {"error": f"HTTP {response.status_code}", "detail": response.text[:200]} except Exception as e: return {"error": str(e)} # ========== 核心函数：AI 助手 ========== def chatbot_response(message, history): """AI 助手主函数""" try: # 构建消息历史 messages = [{"role": "system", "content": SYSTEM_PROMPT}] # 添加对话历史（最近5轮） if history: for item in history[-5:]: if isinstance(item, dict): messages.append(item) elif isinstance(item, (list, tuple)) and len(item) == 2: user_msg, assistant_msg = item messages.append({"role": "user", "content": user_msg}) messages.append({"role": "assistant", "content": assistant_msg}) messages.append({"role": "user", "content": message}) # LLM 调用循环（支持多轮工具调用） tool_calls_log = [] max_iterations = 5 for iteration in range(max_iterations): # 调用 LLM response = client.chat_completion( messages=messages, model="Qwen/Qwen2.5-72B-Instruct:novita", tools=ALL_TOOLS, max_tokens=3000, temperature=0.7, tool_choice="auto" ) choice = response.choices[0] # 检查是否有工具调用 if choice.message.tool_calls: messages.append(choice.message) for tool_call in choice.message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # 记录工具调用 tool_calls_log.append({"name": tool_name, "arguments": tool_args}) # 调用 MCP 工具 tool_result = call_mcp_tool(tool_name, tool_args) # 添加工具结果到消息 messages.append({ "role": "tool", "name": tool_name, "content": json.dumps(tool_result), "tool_call_id": tool_call.id }) continue # 继续下一轮 else: # 无工具调用，返回最终答案 response_text = choice.message.content break # 构建最终响应 final_response = "" # 显示模型信息 final_response += f"
🤖 Model: Qwen/Qwen2.5-72B-Instruct:novita
\n\n" # 显示工具调用日志 if tool_calls_log: final_response += "**🛠️ MCP Tools Used:**\n\n" for i, tool_call in enumerate(tool_calls_log, 1): final_response += f"{i}. `{tool_call['name']}` - {json.dumps(tool_call['arguments'])}\n" final_response += "\n---\n\n" final_response += response_text return final_response except Exception as e: return f"❌ Error: {str(e)}" # ========== Gradio 界面 ========== with gr.Blocks(title="Financial & Market AI Assistant") as demo: gr.Markdown("# 🤖 Financial & Market AI Assistant") gr.Markdown("""
✅ AI Powered by: Qwen/Qwen2.5-72B-Instruct:novita
📊 Data Sources: SEC Financial Reports + Market & Stock Data
""") chat = gr.ChatInterface( fn=chatbot_response, examples=[ "What's Apple's latest revenue and profit?", "Show me NVIDIA's 3-year financial trends", "How is Tesla's stock performing today?", "Compare Microsoft's earnings with its stock price", "Analyze Amazon's cash flow and market cap", ], title="💬 AI Assistant", description="Ask me about company financials, stock prices, or market trends. I'll automatically fetch the data you need!" ) # 启动应用 if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, show_error=True, ssr_mode=False )