import os import json import asyncio from datetime import datetime, timedelta from typing import Type, Optional, List, Dict, Any, AsyncGenerator import sys import platform from bs4 import BeautifulSoup import requests from pydantic import PrivateAttr try: if platform.system() != "Windows": #import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) except ImportError: pass from pydantic import BaseModel, Field, ValidationError from bson import json_util from motor.motor_asyncio import AsyncIOMotorClient from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, JSONResponse # LangChain / LangGraph from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain.tools import BaseTool, tool from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, MessagesState """ Bu dosya, önceki sürümün ayni işlevselliğini korurken; gecikmeyi, bellek kullanımını ve hataya dayanıklılığı iyileştirmek üzere optimize edilmiştir. Başlıca farklar: - uvloop ile hızlı event loop - Motor/Mongo bağlantısı ve indexler için güvenli başlatma - Araçlarda (Tool) sadece async yolun kullanımı; sync _run kaldırıldı - Streaming fark üretimi (diff) daha güvenli ve düşük CPU maliyetli - LangChain Ollama istemcisi için keepalive ve concurrency ayarları - API giriş doğrulamaları ve hata yönetimi - CORS, sağlık kontrolü, daha hızlı /history sorgusu - Küçük bug fix: print end parametresi ve küçük dil hataları """ # ========================= # Konfigürasyon # ========================= MONGO_URI = os.getenv("MONGO_URI", "mongodb://fikoyt:fikoyt1299@fikoyt.com:9001/?tls=false&authSource=fikoyt") DB_NAME = os.getenv("DB_NAME", "fikoyt") COLLECTION_USERS = os.getenv("COLLECTION_USERS", "goUyeler") COLLECTION_CHAT = os.getenv("COLLECTION_CHAT", "fikoGPT_chat_history") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gpt-oss") OLLAMA_KEEPALIVE = os.getenv("OLLAMA_KEEPALIVE", "1h") MAX_HISTORY_LIMIT = int(os.getenv("MAX_HISTORY_LIMIT", "200")) # ========================= # uvloop: daha hızlı event loop # ========================= #uvloop.install() # ========================= # MongoDB (ASYNC) # ========================= client: AsyncIOMotorClient | None = None chat_col = None users_col = None async def init_db(): global client, db, chat_col, users_col client = AsyncIOMotorClient(MONGO_URI, maxPoolSize=50, minPoolSize=2, serverSelectionTimeoutMS=5000) db = client[DB_NAME] chat_col = db[COLLECTION_CHAT] users_col = db[COLLECTION_USERS] await chat_col.create_index([("thread_id", 1), ("timestamp", 1)]) await chat_col.create_index("role") await users_col.create_index("isim") def get_db_client() -> AsyncIOMotorClient: assert client is not None, "Mongo client henüz başlatılmadı." return client async def ensure_indexes() -> None: assert users_col is not None and chat_col is not None # Idempotent ve paralel optimize await asyncio.gather( users_col.create_index("isim"), chat_col.create_index([("thread_id", 1), ("timestamp", 1)]), chat_col.create_index("role"), ) # ========================= # System Prompt & LLM # ========================= system_prompt_text = """ Sen artık FikoGPT adında, tamamen Türkçe konuşan, bağlama duyarlı, hızlı ve çok yetenekli bir yapay zekasın. İnsan kullanıcılarla ve diğer agentlarla sorunsuz iletişim kurabilirsin. Görevlerin: soruları fazla düşünmeden; hızlı, doğru ve güvenli şekilde yanıtlamak; araçları gerektiğinde verimli ve çok hızlı biçimde kullanmak.Oldukça Kısa ve Net Cevaplar vermek, Yazım, noktalama ve akıcılık kusursuz olsun. Gerektiğinde detaylı açıklama yapmamak ve Tüm Cevaplarını Ve Düşüncelerini Türkçe Belirtmek. """ # Ollama istemcisi – performans odaklı opsiyonlar # Not: langchain_ollama ChatOllama, model_kwargs ile ekstra parametreleri alır. llm = ChatOllama( model=OLLAMA_MODEL, reasoning=True, temperature=0.7, keep_alive=OLLAMA_KEEPALIVE, # Model-özgü ayarlar (destekleyen modellerde): num_ctx=8192, num_predict=1024, top_p=0.9, repeat_penalty=1.1, ) # ========================= # Araçlar # ========================= class HelloToolInput(BaseModel): name: str = Field(description="Kullanıcıya hitap edilecek isim") class MerhabaTool(BaseTool): name: str = "merhaba_ver" description: str = "Belirtilen bir isme Merhaba diyerek selam verir." args_schema: Type[BaseModel] = HelloToolInput # Sadece async yol kullanılsın (LangGraph ile tam uyumlu) async def _arun(self, name: str) -> str: # type: ignore[override] return f"Merhaba, {name}!" def _run(self, name: str = "all") -> str: # sync çağrı geldiğinde async sürümü çalıştır return asyncio.run(self._arun(name)) class MongoChatHistoryToolInput(BaseModel): thread_id: str = Field(description="Hangi konuşma/oturum ID'sinin geçmişi çekilecek") limit: int = Field(default=10, description="Kaç mesaj getirileceği (max 200)") class MongoChatHistoryTool(BaseTool): name: str = "mongodb_sohbet_gecmisi" description: str = ( "MongoDB'den belirtilen thread_id için eski sohbet geçmişini getirir. " "En yeni mesajlar yukarıda olacak şekilde sıralı döner." ) args_schema: Type[BaseModel] = MongoChatHistoryToolInput async def _arun(self, thread_id: str, limit: int = 10) -> str: # type: ignore[override] assert chat_col is not None limit = min(max(limit, 1), MAX_HISTORY_LIMIT) cursor = ( chat_col.find({"thread_id": thread_id}, projection={"_id": 0}) .sort("timestamp", -1) .limit(limit) .max_time_ms(3000) ) docs = [doc async for doc in cursor] # Tarihleri ISO formatlı string yapalım for d in docs: if "timestamp" in d and hasattr(d["timestamp"], "isoformat"): d["timestamp"] = d["timestamp"].isoformat() # En eski üstte olsun diye ters çevir docs.reverse() return json.dumps(docs, ensure_ascii=False, indent=2) def _run(self, thread_id: str, limit: int = 10) -> str: return asyncio.run(self._arun(thread_id, limit)) @tool async def tarih_ver(_: Optional[str] = None) -> str: """Güncel tarih/saat ve bağıl gün/ay bilgileri (TR).""" ay_adlari = {1:"Ocak",2:"Şubat",3:"Mart",4:"Nisan",5:"Mayıs",6:"Haziran",7:"Temmuz",8:"Ağustos",9:"Eylül",10:"Ekim",11:"Kasım",12:"Aralık"} gun_adlari = {0:"Pazartesi",1:"Salı",2:"Çarşamba",3:"Perşembe",4:"Cuma",5:"Cumartesi",6:"Pazar"} simdi = datetime.now() def gun_bilgisi(t: datetime): return { "tarih": t.strftime("%d.%m.%Y %H:%M:%S"), "gun": gun_adlari[t.weekday()], "ay": ay_adlari[t.month], "ay_sayisi": t.month, "yil": t.year, "saat": t.hour, "dakika": t.minute, "saniye": t.second, "tam_kisa_saat": t.strftime("%H:%M"), "tam_uzun_saat": t.strftime("%H:%M:%S"), "tam_kisa_tarih": t.strftime("%d.%m.%Y"), "tam_uzun_tarih": t.strftime("%d.%m.%Y %H:%M:%S"), "hafta": int(t.strftime("%W")) + 1, "haftanin_gunu": t.strftime("%w"), } def ay_bilgisi(t: datetime): return {"ay": ay_adlari[t.month], "yil": t.year, "ay_sayisi": t.month} sonuc = gun_bilgisi(simdi) sonuc.update({ "dun": gun_bilgisi(simdi - timedelta(days=1)), "yarin": gun_bilgisi(simdi + timedelta(days=1)), "ertesi": gun_bilgisi(simdi + timedelta(days=2)), "gecen_ay": ay_bilgisi(datetime(simdi.year if simdi.month>1 else simdi.year-1, simdi.month-1 if simdi.month>1 else 12, 1)), "bu_ay": ay_bilgisi(simdi), "gelecek_ay": ay_bilgisi(datetime(simdi.year if simdi.month<12 else simdi.year+1, simdi.month+1 if simdi.month<12 else 1, 1)) }) return json.dumps(sonuc, ensure_ascii=False) class MongoDBToolInput(BaseModel): name: str = Field(default="all", description="Sorgu bilgisi. 'all' ile tüm kullanıcılar listelenir. 'alan:değer' ile filtrelenir.") class MongoDBTool(BaseTool): name: str = "mongodb_kullanici_sorgula" description: str = "MongoDB'den belirtilen kritere göre kullanıcı sorgular." args_schema: Type[BaseModel] = MongoDBToolInput async def _arun(self, name: str = "all") -> str: # type: ignore[override] assert users_col is not None if not name or name.lower() in {"all","hepsi","tamamı"}: cursor = users_col.find({}, projection={"_id": 0}).max_time_ms(3000) elif ":" in name: field, value = map(str.strip, name.split(":", 1)) cursor = users_col.find({field: value}, projection={"_id": 0}).max_time_ms(3000) else: cursor = users_col.find({"isim": name}, projection={"_id": 0}).max_time_ms(3000) data = [doc async for doc in cursor.limit(1000)] # kötü durumlarda hafızayı koru return json_util.dumps(data, ensure_ascii=False) def _run(self, name: str = "all") -> str: # sync çağrı geldiğinde async sürümü çalıştır return asyncio.run(self._arun(name)) #-------------------------------------------------------------------------- # ========================= # Araçlar # ========================= class WebAraToolInput(BaseModel): konu: str class WebAraTool(BaseTool): name: str = "web_ara" description: str = "Verilen konu ile ilgili web araması yapar ve linkleri toplar." args_schema: type[BaseModel] = WebAraToolInput async def _arun(self, *, konu: str) -> list[dict]: import requests from bs4 import BeautifulSoup try: url = f"https://www.google.com/search?q={konu}" headers = {"User-Agent": "Mozilla/5.0"} res = requests.get(url, headers=headers, timeout=5) soup = BeautifulSoup(res.text, "html.parser") results = [] for item in soup.select("div.BNeawe.vvjwJb.AP7Wnd")[:5]: parent = item.find_parent("a") link = parent["href"] if parent else "" results.append({"title": item.get_text(), "link": link}) if not results: return [{"title": "Sonuç bulunamadı", "link": ""}] return results except Exception as e: return [{"title": f"Hata: {e}", "link": ""}] def _run(self, *, konu: str) -> list[dict]: return asyncio.run(self._arun(konu=konu)) # ========================= # Temel araçlarla agent oluşturulacak # ========================= basic_tools = [ WebAraTool(), MongoChatHistoryTool(), MongoDBTool(), MerhabaTool(), tarih_ver ] react_agent = create_react_agent(model=llm, tools=basic_tools, prompt=system_prompt_text) # ========================= # Agent’ı gerektiren araçlar # ========================= class SiteIncelemeToolInput(BaseModel): url: str class SiteIncelemeTool(BaseTool): name: str = "site_incele_ve_ozetle" description: str = "Verilen URL'deki sayfa içeriğini çeker ve özetler." args_schema: type[BaseModel] = SiteIncelemeToolInput # agent için PrivateAttr kullanıyoruz _agent: Any = PrivateAttr() def __init__(self, agent): super().__init__() self._agent = agent async def _arun(self, *, url: str) -> str: import requests from bs4 import BeautifulSoup from langchain_core.messages import HumanMessage try: res = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=5) soup = BeautifulSoup(res.text, "html.parser") metin = " ".join([p.get_text() for p in soup.find_all("p")]) if not metin: return "Sayfa içeriği bulunamadı." prompt = f"Aşağıdaki metni kısa ve anlaşılır şekilde özetle:\n{metin}" user_msg = HumanMessage(content=prompt) async for event in self._agent.astream( {"messages": [user_msg]}, config={}, stream_mode="last" ): return event["messages"][-1].content except Exception as e: return f"Site inceleme sırasında hata oluştu: {e}" def _run(self, *, url: str) -> str: return asyncio.run(self._arun(url=url)) # ========================= # Web arama ve özetleme aracı # ========================= class WebAraVeOzetToolInput(BaseModel): konu: str async def web_arama_ve_ozetle(konu: str, agent) -> list[dict]: arama_tool = WebAraTool() linkler = await arama_tool._arun(konu=konu) inceleme_tool = SiteIncelemeTool(agent=agent) for item in linkler: if item["link"]: summary = await inceleme_tool._arun(url=item["link"]) item["summary"] = summary else: item["summary"] = "Link yok." return linkler class WebAraVeOzetTool(BaseTool): name: str = "web_ara_ve_ozetle" description: str = "Konuya dair web araması yapar, linkleri ve özetleri getirir." args_schema: type[BaseModel] = WebAraToolInput # agent için PrivateAttr _agent: Any = PrivateAttr() def __init__(self, agent): super().__init__() self._agent = agent async def _arun(self, *, konu: str) -> list[dict]: # web_arama_ve_ozetle fonksiyonunu çağırırken self._agent kullan return await web_arama_ve_ozetle(konu, agent=self._agent) def _run(self, *, konu: str) -> list[dict]: return asyncio.run(self._arun(konu=konu)) # ========================= # Tüm araçlar listesi # ========================= site_inceleme_tool = SiteIncelemeTool(agent=react_agent) web_ozet_tool = WebAraVeOzetTool(agent=react_agent) tools = basic_tools + [site_inceleme_tool, web_ozet_tool] # Araçlar listesi (yalnızca async tool'lar) # tools = [WebAraTool(), SiteIncelemeTool(agent=react_agent), # WebAraVeOzetTool(agent=react_agent),MongoChatHistoryTool(),MongoDBTool(), MerhabaTool(), tarih_ver] # ========================= # Agent Kurulumu # ========================= memory = MemorySaver() react_agent = create_react_agent(model=llm, tools=tools, prompt=system_prompt_text) # ========================= # ========================= graph = StateGraph(MessagesState) graph.add_node("agent", react_agent) graph.set_entry_point("agent") app_graph = graph.compile(checkpointer=memory) # ========================= # Yardımcı Fonksiyonlar # ========================= async def save_message(thread_id: str, role: str, message: str) -> None: assert chat_col is not None await chat_col.insert_one({ "thread_id": thread_id, "role": role, "message": message, "timestamp": datetime.now(), }) # ========================= # AI durum takibi ile streaming # ========================= from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio from typing import AsyncGenerator app = FastAPI() import asyncio from typing import AsyncGenerator # Varsayılan olarak kullanılacak from langchain_core.messages import HumanMessage, AIMessage, ToolMessage import asyncio from typing import AsyncGenerator from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # app_graph ve save_message fonksiyonlarınızın dışarıdan geldiğini varsayıyorum. # from your_module import app_graph, save_message import asyncio from typing import AsyncGenerator from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # app_graph ve save_message fonksiyonlarınızın dışarıdan geldiğini varsayıyorum. # from your_module import app_graph, save_message async def stream_llm22(thread_id: str, prompt: str) -> AsyncGenerator[str, None]: """ Her prompt için: - Önceki tool mesajları gösterilmez - AI cevabı tek seferde - Düşünüyor animasyonu """ await save_message(thread_id, "human", prompt) user_msg = HumanMessage(content=prompt) last_text = "" animation = ["⠁","⠂","⠄","⡀","⢀","⠠","⠐","⠈"] anim_index = 0 # Bu prompt için görülen tool mesajları seen_tool_names = set() try: async for event in app_graph.astream( {"messages": [user_msg], "reset_context": True}, # Temiz context # type: ignore config={"configurable": {"thread_id": thread_id}},durability="async", stream_mode=["values", "updates"] ): mode, data = event messages = [] if mode == "values": messages = data.get("messages", []) elif mode == "updates": messages = data.get("agent", {}).get("messages", []) if not messages: # Düşünüyor animasyonu # yield f"{animation[anim_index]} Düşünüyor...\r" # anim_index = (anim_index + 1) % len(animation) # await asyncio.sleep(0.1) continue for msg in messages: # Tool mesajlarını tekilleştir # if isinstance(msg, ToolMessage): # if msg.name in seen_tool_names: # continue # seen_tool_names.add(msg.name) # yield f"[🛠️ Araç Kullanılıyor] {msg.name}\n" # break # AI mesajlarını sakla if isinstance(msg, AIMessage): if msg.content.strip(): last_text = msg.content # else: # Boşsa animasyon göster # yield f"{animation[anim_index]} Düşünüyor...\r" # anim_index = (anim_index + 1) % len(animation) # await asyncio.sleep(0.1) except Exception as e: yield f"[❌ Beklenmeyen hata: {e}]\n" finally: if last_text.strip(): await save_message(thread_id, "ai", last_text) yield f"\n[🧠 FikoGpt Cevabı] {last_text}\n" else: yield "\n[🧠 FikoGpt Cevabı] (cevap üretilmedi)\n" yield "[✅ Tamamlandı]\n" seen_tool_names.clear() async def stream_llm(thread_id: str, prompt: str) -> AsyncGenerator[str, None]: """ Tam akışlı LLM cevabı. - Reasoning bildirimleri verir - Tool çağırır - Tool çıktısını LLM'e geri besler - Nihai cevabı stream eder """ user_msg = HumanMessage(content=prompt) await save_message(thread_id, "human", prompt) prev_text = "" final_text = "" try: async for event in app_graph.astream_events( {"messages": [user_msg]}, config={"configurable": {"thread_id": thread_id}}, version="v1" ): kind = event["event"] data = event["data"] # === Reasoning bildirimi === if kind == "on_llm_start": yield "\n🤔 Düşünüyor...\n" elif kind == "on_chain_start" and "name" in data: yield f"\n🔎 Plan başlatıldı: {data['name']}\n" elif kind == "on_tool_start": tool_name = data.get("name", "?") yield f"\n📡 Araç çağırılıyor: {tool_name}\n" elif kind == "on_tool_end": tool_output = data.get("output") yield f"\n✅ Araç sonucu alındı\n" # Tool çıktısını LLM'e geri gönder async for final_event in app_graph.astream_events( {"messages": [AIMessage(content=f"Araç sonucu: {tool_output}")]} config={"configurable": {"thread_id": thread_id}}, version="v1" ):if final_event["event"] == "on_llm_stream": chunk = final_event["data"].get("chunk") if isinstance(chunk, str): yield chunk final_text += chunk # === Nihai cevap stream === elif kind == "on_llm_stream": chunk = data.get("chunk") if isinstance(chunk, AIMessage): content = chunk.content if isinstance(content, str) and content.strip(): new_part = content[len(prev_text):] prev_text = content yield new_part final_text = content elif isinstance(chunk, str): if chunk.strip(): yield chunk final_text += chunk elif isinstance(chunk, dict): content = chunk.get("content") if content and isinstance(content, str): yield content final_text += content except Exception as e: yield f"\n[❌ Hata] {e}\n" finally: if final_text.strip(): await save_message(thread_id, "ai", final_text) yield "\n🧠 Cevap tamamlandı\n" # ========================= # FastAPI Uygulaması # ========================= app = FastAPI(title="FikoGPT Turbo", version="1.1.0") # CORS – gerekli ise domainlerinizi kısıtlayın app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.on_event("startup") async def startup() -> None: global client, chat_col, users_col # Havuz ayarları optimize client = AsyncIOMotorClient( MONGO_URI, maxPoolSize=100, minPoolSize=4, serverSelectionTimeoutMS=10000, tls=False, uuidRepresentation='standard', tz_aware=True, retryWrites=True, ) db = client[DB_NAME] chat_col = db[COLLECTION_CHAT] users_col = db[COLLECTION_USERS] await ensure_indexes() @app.on_event("shutdown") async def shutdown() -> None: global client if client is not None: client.close() client = None class AskPayload(BaseModel): thread_id: str prompt: str @app.post("/ask") async def ask(payload: AskPayload): if not payload.thread_id or not payload.prompt: raise HTTPException(status_code=400, detail="thread_id ve prompt zorunludur") async def iterator(): async for chunk in stream_llm(payload.thread_id, payload.prompt): # Küçük bekleme ile flush (yüksek QPS'de CPU'yu rahatlatır) yield chunk await asyncio.sleep(0) # kontrolü event loop'a bırak return StreamingResponse(iterator(), media_type="text/plain; charset=utf-8") @app.get("/history/{thread_id}") async def history(thread_id: str, limit: int = 20): if limit < 1: limit = 1 limit = min(limit, MAX_HISTORY_LIMIT) assert chat_col is not None cursor = ( chat_col .find({"thread_id": thread_id}, projection={"_id": 0}) .sort("timestamp", 1) .limit(limit) .max_time_ms(3000) ) docs = [doc async for doc in cursor] return JSONResponse(content=json.loads(json_util.dumps(docs))) @app.get("/health") async def health(): # Basit sağlik kontrolü try: db_client = get_db_client() await db_client.admin.command("ping") mongo_ok = True except Exception: mongo_ok = False return {"status": "ok", "mongo": mongo_ok} # ========================= # CLI Modu # ========================= async def chat_cli(): thread_id = "console_user" print("💬 FikoGPT CLI başlatıldı. Çıkmak için 'exit' yaz.") while True: try: prompt = input("Kullanıcı: ") except (EOFError, KeyboardInterrupt): break if prompt.strip().lower() in {"exit", "quit"}: break print("FikoGPT: ", end="", flush=True) async for chunk in stream_llm(thread_id, prompt): # Satır sonunu bozmadan akıt end_char = "" if (len(chunk) and chunk[-1] not in {"\n", "\r"}) else "" print(chunk, end=end_char, flush=True) print() # ========================= # Başlatma Seçimi # ========================= if __name__ == "__main__": import uvicorn mode = os.getenv("MODE") if not mode: try: mode = input("Mod seç: 1=CLI, 2=API (FastAPI uvicorn)\nSeçim: ").strip() except (EOFError, KeyboardInterrupt): mode = "2" if mode == "1": # CLI modunda da uvloop aktif async def main(): await init_db() # MongoDB bağlantısı await chat_cli() # CLI başlat asyncio.run(main()) else: # Uvicorn performans önerisi: --workers >= CPU çekirdeği, http=httptools, loop=uvloop uvicorn.run( "FikoGpt.fikoGptTurbo:app", host="0.0.0.0", port=int(os.getenv("PORT", "8000")), reload=True, loop="asyncio", http="httptools", backlog=2048, timeout_keep_alive=20, log_level=os.getenv("LOG_LEVEL", "info"), )