Spaces:
Running
Running
File size: 3,840 Bytes
00639e5 3ba1438 00639e5 ca1791d 00639e5 7b37610 00639e5 516fa71 ca1791d 00639e5 ca1791d 00639e5 ca1791d 3ba1438 ca1791d 00639e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | # src/llm_client.py
from abc import ABC, abstractmethod
from typing import Optional
import os
from dotenv import load_dotenv
from langsmith import traceable, get_current_run_tree
load_dotenv()
class LLMClient(ABC):
"""Abstract base class for LLM clients. LLM değişimi için interface."""
@abstractmethod
def generate(
self,
system_prompt: str,
user_prompt: str,
temperature: float = 0.3,
max_tokens: int = 2048
) -> str:
pass
class GeminiClient(LLMClient):
"""Google Gemini LLM client (google-genai SDK)"""
def __init__(
self,
model_name: str = "gemini-2.5-flash",
api_key: Optional[str] = None
):
try:
from google import genai
except ImportError:
raise ImportError(
"google-genai paketi kurulu değil. "
"Kurmak için: pip install google-genai"
)
self.api_key = api_key or os.getenv("GEMINI_API_KEY")
if not self.api_key:
raise ValueError(
"GEMINI_API_KEY bulunamadı. "
".env dosyasına GEMINI_API_KEY=xxx ekle."
)
self.model_name = model_name
self._client = genai.Client(
api_key=self.api_key,
http_options={"api_version": "v1beta"}
)
print(f"Gemini client initialized: {model_name}")
@traceable(name="Gemini Report Generation", run_type="llm",
metadata={"model": "gemini-2.5-flash", "temperature": 0.3})
def generate(
self,
system_prompt: str,
user_prompt: str,
temperature: float = 0.3,
max_tokens: int = 2048,
min_response_chars: int = 800,
max_retries: int = 2
) -> str:
from google.genai import types
full_prompt = f"{system_prompt}\n\n{user_prompt}"
for attempt in range(max_retries + 1):
try:
response = self._client.models.generate_content(
model=self.model_name,
contents=full_prompt,
config=types.GenerateContentConfig(
temperature=temperature,
max_output_tokens=max_tokens,
thinking_config=types.ThinkingConfig(
thinking_budget=0
),
)
)
text = response.text
if len(text) >= min_response_chars or attempt == max_retries:
rt = get_current_run_tree()
if rt and response.usage_metadata:
um = response.usage_metadata
rt.add_metadata({
"usage": {
"input_tokens": um.prompt_token_count,
"output_tokens": um.candidates_token_count,
"total_tokens": um.total_token_count,
}
})
return text
print(f" → Short response ({len(text)} chars), retrying ({attempt + 1}/{max_retries})...")
except Exception as e:
if attempt == max_retries:
raise RuntimeError(f"Gemini generation failed: {e}")
print(f" → Generation error, retrying ({attempt + 1}/{max_retries}): {e}")
raise RuntimeError("Gemini generation failed after retries")
# Test
if __name__ == "__main__":
client = GeminiClient()
response = client.generate(
system_prompt="You are a helpful medical AI assistant.",
user_prompt="Briefly explain what a kidney stone is in 2 sentences.",
temperature=0.3
)
print("\nGemini Response:")
print(response)
|