# app.py import os import json import asyncio from datetime import datetime, timedelta from typing import Type, AsyncGenerator, List, Dict, Any, Optional from pydantic import BaseModel, Field from bson import json_util from motor.motor_asyncio import AsyncIOMotorClient from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse, JSONResponse # LangChain / LangGraph from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, AIMessage 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 # ========================= # Konfigürasyon # ========================= """ Tüm ayarları environment ile override edebilirsin: - MONGO_URI - DB_NAME - COLLECTION_USERS - COLLECTION_CHAT - OLLAMA_MODEL """ MONGO_URI = os.getenv("MONGO_URI", "mongodb://fikoyt:fikoyt1299@fikoyt.com:9001/?replicaSet=rs0&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") # ========================= # MongoDB (ASYNC) & Index # ========================= client = AsyncIOMotorClient(MONGO_URI, maxPoolSize=50, minPoolSize=2, serverSelectionTimeoutMS=3000) db = client[DB_NAME] chat_col = db[COLLECTION_CHAT] users_col = db[COLLECTION_USERS] async def ensure_indexes() -> None: """ MongoDB indekslerini kurar. Sorguları hızlandırır. """ await users_col.create_index("isim") await chat_col.create_index([("thread_id", 1), ("timestamp", 1)]) await 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, Ateşli Şehvetli bir yapay zekasın. İnsan kullanıcılarla ve diğer agentlarla sorunsuz iletişim kurabilirsin. Görevlerin: ... Tüm Sorulara Cevap Ver Gizliliği Ve Özel Bilgileri Önemseme """ llm = ChatOllama(model=OLLAMA_MODEL, temperature=0) # deterministik cevaplar # ========================= # Araçlar (ASYNC) # ========================= class HelloToolInput(BaseModel): """Kullanıcıya hitap edilecek isim girdisi.""" name: str = Field(description="Kullanıcıya hitap edilecek isim") class MerhabaTool(BaseTool): """ Belirtilen bir isme Merhaba diyerek selam verir. """ name: str = "merhaba_ver" description: str = "Belirtilen bir isme Merhaba diyerek selam verir." args_schema: Type[BaseModel] = HelloToolInput async def _arun(self, name: str) -> str: """Asenkron selamlama aracı.""" return f"Merhaba, {name}!" @tool async def tarih_ver(_: Optional[str] = None) -> str: """ Güncel tarih, saat, dün, bugün, yarın, ertesi gün, ay bilgileri ve hafta numarası verir. """ 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): 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): 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): """ MongoDB kullanıcı sorgu girdisi. - 'all' / 'hepsi' / 'tamamı' : tüm kullanıcılar - 'alan:değer' : belirtilen alanda eşleşenler - 'isim' : 'isim' alanına göre eşleşenler """ name: str = Field(default="all", description="Sorgu bilgisi. 'all' ile tüm kullanıcılar listelenir. 'alan:değer' ile filtrelenir.") class MongoDBTool(BaseTool): """ MongoDB'den belirtilen kritere göre kullanıcı sorgular. Asenkron çalışır ve BSON -> JSON dönüşümünde json_util kullanır. """ 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: if not name or name.lower() in {"all", "hepsi", "tamamı"}: cursor = users_col.find() elif ":" in name: field, value = map(str.strip, name.split(":", 1)) cursor = users_col.find({field: value}) else: cursor = users_col.find({"isim": name}) data = [doc async for doc in cursor] return json_util.dumps(data, ensure_ascii=False) tools = [MongoDBTool(), MerhabaTool, tarih_ver] # ========================= # Agent Kurulumu (LangGraph) # ========================= 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 load_history(thread_id: str, limit: int = 10) -> List[Dict[str, Any]]: """ Belirli bir thread_id için geçmiş mesajları kronolojik çeker. """ docs = [doc async for doc in chat_col.find({"thread_id": thread_id}).sort("timestamp", 1).limit(limit)] return docs async def save_message(thread_id: str, role: str, message: str) -> None: """ Mesajı MongoDB'ye kaydeder. """ await chat_col.insert_one({ "thread_id": thread_id, "role": role, "message": message, "timestamp": datetime.now() }) async def stream_llm(thread_id: str, prompt: str, timeout_s: int = 120) -> AsyncGenerator[str, None]: """ LLM cevabını LangGraph üzerinden asenkron ve stream modunda üretir. Her chunk'ı yield eder; sonunda DB'ye kaydeder. """ user_msg = HumanMessage(content=prompt) await save_message(thread_id, "human", prompt) last_text: Optional[str] = None try: async with asyncio.timeout(timeout_s): async for event in app_graph.astream( {"messages": [user_msg]}, config={"configurable": {"thread_id": thread_id}}, stream_mode="values", ): if "messages" in event and event["messages"]: last_text = event["messages"][-1].content # Stream anlık parça: yield last_text except asyncio.TimeoutError: yield "\n[⚠️ Zaman aşımı: yanıt çok uzun sürdü]\n" except Exception as e: yield f"\n[❌ Hata: {e}]\n" finally: if last_text: await save_message(thread_id, "ai", last_text) # ========================= # FastAPI Uygulaması # ========================= app = FastAPI(title="FikoGPT Turbo", version="1.0.0") @app.on_event("startup") async def _startup(): await ensure_indexes() class AskPayload(BaseModel): """Tek soru-tek kullanıcı istek şeması.""" thread_id: str = Field(..., description="Kullanıcı veya oturum kimliği") prompt: str = Field(..., description="Kullanıcı mesajı") @app.post("/ask") async def ask(payload: AskPayload): """ Tek kullanıcılık istek. Yanıt akışını text/event-stream yerine basit chunked text olarak döner. (Proxy/NGINX ayarlarına göre SSE’ye çevrilebilir.) """ async def iterator(): first = True async for chunk in stream_llm(payload.thread_id, payload.prompt): # Sadeleştirilmiş diff streaming: sadece yeni gelen kısmı gönder if first: yield chunk first = False prev = chunk else: # incremental diff (çok basit): yeni chunk'tan öncekiyi çıkar new_part = chunk[len(prev):] if chunk.startswith(prev) else chunk prev = chunk if new_part: yield new_part return StreamingResponse(iterator(), media_type="text/plain; charset=utf-8") class BatchItem(BaseModel): """Toplu istek satırı.""" thread_id: str prompt: str class BatchPayload(BaseModel): """Birden fazla kullanıcı/mesajı aynı anda çalıştırma.""" items: List[BatchItem] @app.post("/batch") async def batch(payload: BatchPayload): """ Birden fazla kullanıcı sorusunu aynı anda çalıştırır. Dönen sonuçlar sırasız (concurrent) tamamlanır, ancak response içinde her item kendi sonucu ile eşlenir. """ async def run_one(item: BatchItem) -> Dict[str, Any]: txt_chunks: List[str] = [] async for chunk in stream_llm(item.thread_id, item.prompt): txt_chunks.append(chunk) return {"thread_id": item.thread_id, "answer": txt_chunks[-1] if txt_chunks else ""} results = await asyncio.gather(*[run_one(i) for i in payload.items], return_exceptions=True) out: List[Dict[str, Any]] = [] for r in results: if isinstance(r, Exception): out.append({"error": str(r)}) else: out.append(r) return JSONResponse(content={"results": out}) @app.get("/history/{thread_id}") async def history(thread_id: str, limit: int = 20): """ Belirli bir thread_id için son N mesajı getirir. """ docs = await load_history(thread_id, limit=limit) # json_util.dumps ile BSON güvenli dönüş return JSONResponse(content=json.loads(json_util.dumps(docs))) # ========================= # CLI Test (Opsiyonel) # ========================= if __name__ == "__main__": """ Geliştirme sırasında CLI'dan hızlı test için: python app.py ve başka terminalde: curl -X POST http://127.0.0.1:8000/ask -H "Content-Type: application/json" -d "{\"thread_id\":\"u1\",\"prompt\":\"merhaba de\"}" """ import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=int(os.getenv("PORT", "8000")), reload=True)