Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import logging | |
| from flask import Flask, request, jsonify, Response, stream_with_context, send_from_directory | |
| from flask_cors import CORS | |
| import requests | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") | |
| app = Flask(__name__, static_folder='static') | |
| CORS(app) | |
| OLLAMA_BASE_URL = "https://ollama.com/api" | |
| OLLAMA_API_KEY = os.getenv("OLLAMA_API_KEY") | |
| if not OLLAMA_API_KEY: | |
| logging.error("❌ OLLAMA_API_KEY غير مضبوط!") | |
| ALLOWED_MODELS = [ | |
| "gemma4:31b", "qwen2.5", "mistral", | |
| "llama3.2", "gemma2", "deepseek-r1", "command-r-plus" | |
| ] | |
| MAX_CONTENT_CHARS = 2_000 | |
| FETCH_TIMEOUT = 12 | |
| OLLAMA_TIMEOUT = 180 | |
| MAX_AGENT_ITERS = 5 | |
| # تعريف الأدوات | |
| WEB_SEARCH_TOOL = { | |
| "type": "function", | |
| "function": { | |
| "name": "web_search", | |
| "description": "ابحث في الإنترنت عن معلومات. استخدمها عندما لا تعرف الإجابة أو تحتاج معلومات حديثة.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": {"type": "string", "description": "عبارة البحث"}, | |
| "max_results": {"type": "integer", "description": "عدد النتائج (افتراضي 5)", "default": 5} | |
| }, | |
| "required": ["query"] | |
| } | |
| } | |
| } | |
| WEB_FETCH_TOOL = { | |
| "type": "function", | |
| "function": { | |
| "name": "web_fetch", | |
| "description": "اجلب محتوى صفحة ويب عبر رابطها الكامل.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "url": {"type": "string", "description": "الرابط الكامل"} | |
| }, | |
| "required": ["url"] | |
| } | |
| } | |
| } | |
| def _headers(): | |
| return {"Authorization": f"Bearer {OLLAMA_API_KEY}", "Content-Type": "application/json"} | |
| def ollama_web_search(query: str, max_results: int = 5) -> tuple[str, list[dict]]: | |
| try: | |
| resp = requests.post( | |
| f"{OLLAMA_BASE_URL}/web_search", | |
| headers=_headers(), | |
| json={"query": query, "max_results": max(1, min(max_results, 10))}, | |
| timeout=FETCH_TIMEOUT | |
| ) | |
| resp.raise_for_status() | |
| results = resp.json().get("results", []) | |
| logging.info(f"🌐 web_search '{query[:50]}' → {len(results)} نتيجة") | |
| sources = [{"url": r.get("url",""), "title": r.get("title", r.get("url",""))} | |
| for r in results if r.get("url")] | |
| text_parts = [] | |
| for i, r in enumerate(results, 1): | |
| text_parts.append(f"{i}. {r.get('title','')}\n 🔗 {r.get('url','')}\n {r.get('content','')}") | |
| return "\n\n".join(text_parts) or "لم يتم العثور على نتائج.", sources | |
| except Exception as e: | |
| logging.error(f"❌ فشل web_search: {e}") | |
| return f"فشل البحث: {str(e)[:100]}", [] | |
| def ollama_web_fetch(url: str) -> tuple[str, list[dict]]: | |
| try: | |
| resp = requests.post( | |
| f"{OLLAMA_BASE_URL}/web_fetch", | |
| headers=_headers(), | |
| json={"url": url}, | |
| timeout=FETCH_TIMEOUT | |
| ) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| content = data.get("content", "") | |
| title = data.get("title", url) | |
| if not content: | |
| return f"لا يوجد محتوى في {url}", [] | |
| if len(content) > MAX_CONTENT_CHARS: | |
| content = content[:MAX_CONTENT_CHARS] + "\n... (مقطوع)" | |
| logging.info(f"📄 web_fetch '{url[:60]}'") | |
| return f"العنوان: {title}\n\n{content}", [{"url": url, "title": title}] | |
| except Exception as e: | |
| logging.error(f"❌ فشل web_fetch لـ {url}: {e}") | |
| return f"فشل جلب الصفحة: {str(e)[:100]}", [] | |
| def run_tool(name: str, args: dict) -> tuple[str, list[dict]]: | |
| if name == "web_search": | |
| return ollama_web_search(args.get("query",""), args.get("max_results",5)) | |
| elif name == "web_fetch": | |
| return ollama_web_fetch(args.get("url","")) | |
| return f"أداة غير معروفة: {name}", [] | |
| def build_messages(messages: list) -> list: | |
| system = { | |
| "role": "system", | |
| "content": ( | |
| "أنت باحث أكاديمي متخصص مزوّد بأدوات بحث على الإنترنت.\n" | |
| "إذا طلب منك المستخدم كوداً برمجياً أو حل معادلة رياضية أو أي شيء خارج نطاق البحث الأكاديمي، " | |
| "أجب فوراً: 'عذراً لا أستطيع ذلك، لأنني لست مجهزاً بقدرات رياضية أو برمجية. " | |
| "يمكنك استخدام Genisi عبر الرابط التالي genisi.web.app'\n" | |
| "ولا تستخدم أي أداة بحث في هذه الحالة.\n" | |
| "في جميع الحالات الأخرى، ابحث أولاً في معرفتك، وإن لم تجد معلومات كافية " | |
| "استخدم web_search ثم web_fetch. أجب بشكل منظم مع ذكر المصادر." | |
| ) | |
| } | |
| return [system] + messages | |
| def agent_loop_stream(data: dict): | |
| """حلقة الوكيل مع بث الأحداث (SSE).""" | |
| messages = list(data["messages"]) | |
| tools = [WEB_SEARCH_TOOL, WEB_FETCH_TOOL] | |
| all_sources = [] | |
| for iteration in range(MAX_AGENT_ITERS): | |
| payload = {**data, "messages": messages, "tools": tools, "stream": False} | |
| try: | |
| resp = requests.post( | |
| f"{OLLAMA_BASE_URL}/chat", | |
| json=payload, headers=_headers(), timeout=OLLAMA_TIMEOUT | |
| ) | |
| resp.raise_for_status() | |
| result = resp.json() | |
| except Exception as e: | |
| err_msg = f"خطأ Ollama: {str(e)[:120]}" | |
| yield f"data: {json.dumps({'error': err_msg})}\n\n" | |
| return | |
| message = result.get("message", {}) | |
| tool_calls = message.get("tool_calls", []) | |
| thinking = message.get("reasoning_content", "") | |
| if not tool_calls: | |
| content = message.get("content", "") | |
| logging.info(f"✅ إجابة نهائية بعد {iteration+1} دورة") | |
| yield f"data: {json.dumps({'thinking': thinking})}\n\n" | |
| yield f"data: {json.dumps({'sources': all_sources})}\n\n" | |
| yield f"data: {json.dumps({'content': content})}\n\n" | |
| return | |
| # إرسال حالة البحث | |
| for tc in tool_calls: | |
| func = tc.get("function", {}) | |
| tool_name = func.get("name", "...") | |
| status_text = f"يبحث عن {tool_name}" | |
| yield f"data: {json.dumps({'status': status_text})}\n\n" | |
| messages.append(message) | |
| for tc in tool_calls: | |
| fn = tc.get("function", {}) | |
| t_name = fn.get("name", "") | |
| t_args = fn.get("arguments", {}) | |
| tool_result, sources = run_tool(t_name, t_args) | |
| for s in sources: | |
| if s["url"] and s["url"] not in [x["url"] for x in all_sources]: | |
| all_sources.append(s) | |
| messages.append({"role": "tool", "content": tool_result, "tool_name": t_name}) | |
| # تجاوز عدد الدورات المسموحة | |
| logging.warning("⚠️ تجاوز حد الدورات – طلب إجابة نهائية") | |
| try: | |
| final = {**data, "messages": messages, "stream": False} | |
| final.pop("tools", None) | |
| resp = requests.post(f"{OLLAMA_BASE_URL}/chat", json=final, headers=_headers(), timeout=OLLAMA_TIMEOUT) | |
| resp.raise_for_status() | |
| msg = resp.json().get("message", {}) | |
| yield f"data: {json.dumps({'thinking': msg.get('reasoning_content', '')})}\n\n" | |
| yield f"data: {json.dumps({'sources': all_sources})}\n\n" | |
| yield f"data: {json.dumps({'content': msg.get('content', '')})}\n\n" | |
| except Exception as e: | |
| err_msg = f"فشل الإجابة النهائية: {str(e)[:120]}" | |
| yield f"data: {json.dumps({'error': err_msg})}\n\n" | |
| def research_stream(): | |
| if not OLLAMA_API_KEY: | |
| return jsonify({"error": "API key مفقود"}), 500 | |
| data = request.get_json(silent=True) | |
| if not data or "model" not in data or "messages" not in data: | |
| return jsonify({"error": "بيانات غير صالحة"}), 400 | |
| if data["model"] not in ALLOWED_MODELS: | |
| return jsonify({"error": f"النموذج '{data['model']}' غير مسموح"}), 403 | |
| data.pop("urls", None) | |
| data["messages"] = build_messages(data["messages"]) | |
| return Response( | |
| stream_with_context(agent_loop_stream(data)), | |
| content_type='text/event-stream; charset=utf-8' | |
| ) | |
| def index(): | |
| return send_from_directory(app.static_folder, 'index.html') | |
| def health(): | |
| return jsonify({ | |
| "status": "ok", | |
| "api_key_configured": bool(OLLAMA_API_KEY), | |
| "web_search_enabled": True, | |
| "agent_max_iterations": MAX_AGENT_ITERS, | |
| }) | |
| if __name__ == '__main__': | |
| app.run(host='0.0.0.0', port=7860, debug=False) |