Testapi / app.py
bahi-bh's picture
Update app.py
8cbefa0 verified
Raw History Blame Contribute Delete
90.9 kB
#!/usr/bin/env python3
"""
gemini-web2api + Zep Cloud Memory โ€” Professional Agent Edition
===============================================================
ู†ุณุฎุฉ ุงุญุชุฑุงููŠุฉ ูƒุงู…ู„ุฉ ู…ุน:
- โœ… Zep Cloud: ุฐุงูƒุฑุฉ ุฏุงุฆู…ุฉ ุนุจุฑ ูƒู„ ุงู„ุฌู„ุณุงุช (ุญู‚ุงุฆู‚ + ู…ู„ุฎุต + ุจุญุซ ุฏู„ุงู„ูŠ)
- โœ… Agent Loop ุญู‚ูŠู‚ูŠ (ReAct: Reason โ†’ Act โ†’ Observe โ†’ Repeat)
- โœ… Tool Calling ู…ูˆุซูˆู‚ ู…ุน JSON parser ู…ุชุนุฏุฏ ุงู„ู…ุณุชูˆูŠุงุช
- โœ… Streaming ุญู‚ูŠู‚ูŠ ู…ุน Tool Calls
- โœ… SQLite Cache + Rate Limiting
- โœ… Retry Logic + Fallback ู„ู„ู†ู…ูˆุฐุฌ ุงู„ุงูุชุฑุงุถูŠ
- โœ… Gradio UI ู…ุญุณู‘ู†
ุงู„ู…ุชุทู„ุจุงุช:
pip install fastapi uvicorn aiosqlite httpx gradio zep-cloud pydantic
ู…ุชุบูŠุฑุงุช ุงู„ุจูŠุฆุฉ:
ZEP_API_KEY - ู…ูุชุงุญ Zep Cloud (ู…ุทู„ูˆุจ ู„ู„ุฐุงูƒุฑุฉ ุงู„ุฏุงุฆู…ุฉ)
API_KEYS - ู…ูุงุชูŠุญ API ู…ูุตูˆู„ุฉ ุจูุงุตู„ุฉ (ุงูุชุฑุงุถูŠ: sk-gemini)
GEMINI_COOKIE - ูƒูˆูƒูŠ Gemini (ุงุฎุชูŠุงุฑูŠุŒ ูŠูุญุณู‘ู† ุงู„ุฌูˆุฏุฉ)
HTTPS_PROXY - ุจุฑูˆูƒุณูŠ (ุงุฎุชูŠุงุฑูŠ)
GEMINI_BL - Gemini BL header (ุงุฎุชูŠุงุฑูŠ)
"""
import asyncio
import json
import time
import hashlib
import re
import os
import ssl
import sys
import uuid
import argparse
import urllib.parse
import urllib.request
from typing import Optional, List, Dict, Any, Tuple, Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import aiosqlite
import gradio as gr
try:
import httpx
HAS_HTTPX = True
except ImportError:
HAS_HTTPX = False
try:
from zep_cloud.client import AsyncZep
from zep_cloud.types import Message as ZepMessage
HAS_ZEP = True
except ImportError:
HAS_ZEP = False
__version__ = "3.0.0-zep"
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Configuration โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def load_config() -> dict:
defaults = {
# โ”€โ”€โ”€ Server
"port": 7860,
"host": "0.0.0.0",
"log_requests": True,
# โ”€โ”€โ”€ Gemini
"gemini_bl": "boq_assistant-bard-web-server_20260525.09_p0",
"default_model": "gemini-3.5-flash",
"cookie_file": None,
"proxy": None,
"retry_attempts": 3,
"retry_delay_sec": 2,
"request_timeout_sec": 180,
# โ”€โ”€โ”€ API
"api_keys": ["sk-gemini"],
"rate_limit_per_minute": 20,
# โ”€โ”€โ”€ Cache
"cache_ttl_seconds": 3600,
"db_path": "cache.db",
"cleanup_interval_seconds": 300,
# โ”€โ”€โ”€ Agent Loop
"agent_max_iterations": 10,
"agent_max_tokens_per_step": 4096,
# โ”€โ”€โ”€ Zep Cloud Memory
"zep_api_key": os.environ.get("ZEP_API_KEY", ""),
"zep_memory_type": "perpetual", # perpetual | summary_retriever
"zep_min_fact_rating": 0.3, # ุญุฏ ุฃุฏู†ู‰ ู„ุฌูˆุฏุฉ ุงู„ุญู‚ุงุฆู‚
"zep_recent_messages": 20, # ุนุฏุฏ ุงู„ุฑุณุงุฆู„ ุงู„ุฃุฎูŠุฑุฉ ููŠ ุงู„ุณูŠุงู‚
"zep_search_limit": 5, # ู†ุชุงุฆุฌ ุงู„ุจุญุซ ุงู„ุฏู„ุงู„ูŠ
"zep_save_system_msgs": False, # ู‡ู„ ุชุญูุธ ุฑุณุงุฆู„ ุงู„ู†ุธุงู…ุŸ
}
# ุชุญู…ูŠู„ ู…ู„ู ุงู„ุฅุนุฏุงุฏุงุช ุฅุฐุง ูˆูุฌุฏ
for p in ["./config.json", os.path.expanduser("~/.config/gemini-web2api/config.json")]:
if os.path.exists(p):
try:
with open(p) as f:
defaults.update(json.load(f))
print(f"[Config] Loaded: {p}")
except Exception as e:
print(f"[Config] Error: {e}")
break
# ู…ุชุบูŠุฑุงุช ุงู„ุจูŠุฆุฉ ุชูู„ุบูŠ ุงู„ุฅุนุฏุงุฏุงุช
if os.environ.get("API_KEYS"):
defaults["api_keys"] = [k.strip() for k in os.environ["API_KEYS"].split(",") if k.strip()]
if os.environ.get("GEMINI_COOKIE"):
defaults["_cookie_inline"] = os.environ["GEMINI_COOKIE"]
if os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY"):
defaults["proxy"] = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
if os.environ.get("GEMINI_BL"):
defaults["gemini_bl"] = os.environ["GEMINI_BL"]
if os.environ.get("ZEP_API_KEY"):
defaults["zep_api_key"] = os.environ["ZEP_API_KEY"]
return defaults
CONFIG = load_config()
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Models โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
MODELS = {
"gemini-3.7-flash": {
"mode": 1, "think": 4,
"desc": "Fast general-purpose model (~12k chars)",
},
"gemini-3.6-flash": {
"mode": 1, "think": 4,
"desc": "All-around model (Gemini 3.6 Flash)",
},
"gemini-3.5-flash": {
"mode": 1, "think": 4,
"desc": "Standard flash model",
},
"gemini-3.5-flash-thinking": {
"mode": 2, "think": 0,
"desc": "Deep thinking mode (~20k chars)",
},
"gemini-3.1-pro": {
"mode": 3, "think": 4,
"desc": "Pro model",
},
"gemini-auto": {
"mode": 4, "think": 4,
"desc": "Auto model selection",
},
"gemini-3.5-flash-thinking-lite": {
"mode": 5, "think": 0,
"desc": "Dynamic thinking (~15k chars)",
},
"gemini-flash-lite": {
"mode": 6, "think": 4,
"desc": "Lightweight fast model (~10k chars)",
},
}
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Pydantic Schemas โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
class ChatMessage(BaseModel):
role: str
content: Any
tool_calls: Optional[List[dict]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
class ChatCompletionRequest(BaseModel):
model: str = "gemini-3.5-flash"
messages: List[ChatMessage]
temperature: Optional[float] = 0.7
max_tokens: Optional[int] = 32000
stream: Optional[bool] = False
tools: Optional[List[dict]] = None
tool_choice: Optional[Any] = "auto"
user: Optional[str] = "default"
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Logging โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def log(msg: str, level: str = "INFO"):
if CONFIG.get("log_requests"):
icons = {
"INFO": "โ„น๏ธ", "AGENT": "๐Ÿค–", "TOOL": "๐Ÿ”ง",
"ERROR": "โŒ", "WARN": "โš ๏ธ", "MEM": "๐Ÿง ",
}
icon = icons.get(level, "โ€ข")
sys.stderr.write(f"[{time.strftime('%H:%M:%S')}] {icon} {msg}\n")
sys.stderr.flush()
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Services (Cache + Rate Limit) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@dataclass
class RateLimitEntry:
timestamps: List[float] = field(default_factory=list)
last_cleanup: float = field(default_factory=time.time)
class Services:
def __init__(self):
self._rate_limits: Dict[str, RateLimitEntry] = {}
self._lock = asyncio.Lock()
self._cleanup_task: Optional[asyncio.Task] = None
async def init_db(self):
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
val TEXT,
exp REAL
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_exp ON cache(exp)")
await db.commit()
async def start_cleanup(self):
async def loop():
while True:
await asyncio.sleep(CONFIG["cleanup_interval_seconds"])
await self._cleanup()
self._cleanup_task = asyncio.create_task(loop())
async def _cleanup(self):
now = time.time()
cutoff = now - 3600
async with self._lock:
expired = [ip for ip, e in self._rate_limits.items() if e.last_cleanup < cutoff]
for ip in expired:
del self._rate_limits[ip]
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute("DELETE FROM cache WHERE exp < ?", (now,))
await db.commit()
except Exception:
pass
async def check_rate_limit(self, client_id: str) -> Tuple[bool, int]:
now = time.time()
cutoff = now - 60
async with self._lock:
if client_id not in self._rate_limits:
self._rate_limits[client_id] = RateLimitEntry()
entry = self._rate_limits[client_id]
entry.timestamps = [t for t in entry.timestamps if t > cutoff]
entry.last_cleanup = now
limit = CONFIG["rate_limit_per_minute"]
if len(entry.timestamps) >= limit:
wait = 60 - (now - min(entry.timestamps))
return False, int(wait) + 1
entry.timestamps.append(now)
return True, 0
async def get_cache(self, key: str) -> Optional[str]:
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
async with db.execute(
"SELECT val FROM cache WHERE key=? AND exp>?", (key, time.time())
) as cur:
row = await cur.fetchone()
return row[0] if row else None
except Exception:
return None
async def set_cache(self, key: str, value: str):
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute(
"INSERT OR REPLACE INTO cache VALUES (?, ?, ?)",
(key, value, time.time() + CONFIG["cache_ttl_seconds"])
)
await db.commit()
except Exception:
pass
services = Services()
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Zep Cloud Memory Manager โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
class ZepMemoryManager:
"""
ู…ุฏูŠุฑ ุงู„ุฐุงูƒุฑุฉ ุงู„ุฏุงุฆู…ุฉ ุนุจุฑ Zep Cloud.
ุงู„ู…ูŠุฒุงุช:
โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
โ€ข ูƒู„ user_id โ†’ ุฌู„ุณุฉ (session) ุซุงุจุชุฉ ู„ุง ุชูู…ุณุญ ุจูŠู† ุงู„ุฏุฑุฏุดุงุช
โ€ข ูŠุณุชุฑุฌุน: ุงู„ุญู‚ุงุฆู‚ ุงู„ู…ุณุชุฎุฑุฌุฉ + ู…ู„ุฎุต ุงู„ู…ุญุงุฏุซุฉ + ุขุฎุฑ N ุฑุณุงู„ุฉ
โ€ข ูŠุจุญุซ ุฏู„ุงู„ูŠุงู‹ ููŠ ูƒุงู…ู„ ุชุงุฑูŠุฎ ุงู„ู…ุญุงุฏุซุฉ ุจู†ุงุกู‹ ุนู„ู‰ ุงู„ุณุคุงู„ ุงู„ุญุงู„ูŠ
โ€ข ูŠุญูุธ ูƒู„ ุฑุณุงู„ุฉ (user + assistant + tool) ุชู„ู‚ุงุฆูŠุงู‹ ุจุนุฏ ูƒู„ ุฑุฏ
โ€ข ูŠุชุนุงู…ู„ ู…ุน ุชุบูŠูŠุฑ ุงู„ู…ูˆุถูˆุน: ูƒู„ ุฑุณุงู„ุฉ ู…ุฑุชุจุทุฉ ุจู€ timestamp ูˆุณูŠุงู‚
"""
def __init__(self):
self.client: Optional[AsyncZep] = None
self._enabled = False
self._init_error: str = ""
async def init(self):
"""ุชู‡ูŠุฆุฉ ุนู…ูŠู„ Zep Cloud."""
if not HAS_ZEP:
self._init_error = "zep-cloud not installed. Run: pip install zep-cloud"
log(self._init_error, "WARN")
return
api_key = CONFIG.get("zep_api_key", "").strip()
if not api_key:
self._init_error = "ZEP_API_KEY not set โ€” persistent memory disabled"
log(self._init_error, "WARN")
return
try:
self.client = AsyncZep(api_key=api_key)
# ุงุฎุชุจุงุฑ ุงู„ุงุชุตุงู„
await self.client.user.list(page_number=1, page_size=1)
self._enabled = True
log("โœ… Zep Cloud memory connected and ready", "MEM")
except Exception as e:
self._init_error = f"Zep Cloud connection failed: {e}"
log(self._init_error, "ERROR")
@property
def enabled(self) -> bool:
return self._enabled and self.client is not None
# โ”€โ”€โ”€ Session Management โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def _session_id(self, user_id: str) -> str:
"""
session_id ุซุงุจุช ูˆู…ุจู†ูŠ ุนู„ู‰ user_id.
ู„ุง ูŠุชุบูŠุฑ ุจูŠู† ุงู„ุฌู„ุณุงุช โ†’ ู‡ุฐุง ู‡ูˆ ู…ูุชุงุญ ุงู„ุงุณุชู…ุฑุงุฑูŠุฉ.
"""
safe = re.sub(r'[^a-zA-Z0-9_-]', '_', user_id)
return f"agent_{safe}"
async def ensure_user_session(self, user_id: str) -> bool:
"""
ูŠู†ุดุฆ ุงู„ู…ุณุชุฎุฏู… ูˆุงู„ุฌู„ุณุฉ ููŠ Zep ุฅู† ู„ู… ุชูƒูˆู†ุง ู…ูˆุฌูˆุฏูŽูŠู†.
ูŠุนูŠุฏ True ุนู†ุฏ ุงู„ู†ุฌุงุญ.
"""
if not self.enabled:
return False
try:
session_id = self._session_id(user_id)
# ุฅู†ุดุงุก ุงู„ู…ุณุชุฎุฏู… (ูŠุชุฌุงู‡ู„ ุงู„ุฎุทุฃ ุฅุฐุง ูƒุงู† ู…ูˆุฌูˆุฏุงู‹)
try:
await self.client.user.add(
user_id=user_id,
metadata={
"source": "gemini-web2api",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
}
)
log(f"Created Zep user: {user_id}", "MEM")
except Exception:
pass # ุงู„ู…ุณุชุฎุฏู… ู…ูˆุฌูˆุฏ ุจุงู„ูุนู„
# ุฅู†ุดุงุก ุงู„ุฌู„ุณุฉ (ูŠุชุฌุงู‡ู„ ุงู„ุฎุทุฃ ุฅุฐุง ูƒุงู†ุช ู…ูˆุฌูˆุฏุฉ)
try:
await self.client.memory.add_session(
session_id=session_id,
user_id=user_id,
metadata={
"agent_version": __version__,
"model": CONFIG.get("default_model"),
}
)
log(f"Created Zep session: {session_id}", "MEM")
except Exception:
pass # ุงู„ุฌู„ุณุฉ ู…ูˆุฌูˆุฏุฉ ุจุงู„ูุนู„
return True
except Exception as e:
log(f"ensure_user_session error: {e}", "WARN")
return False
# โ”€โ”€โ”€ Memory Retrieval โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
async def get_full_context(self, user_id: str, current_query: str = "") -> str:
"""
ูŠุณุชุฑุฌุน ุงู„ุณูŠุงู‚ ุงู„ูƒุงู…ู„ ู…ู† Zep ูˆูŠูุนูŠุฏู‡ ูƒู†ุต ู…ู†ุณูŽู‘ู‚.
ูŠุดู…ู„:
1. ุงู„ุญู‚ุงุฆู‚ ุงู„ู…ุณุชุฎุฑุฌุฉ ุชู„ู‚ุงุฆูŠุงู‹ (ู…ุซู„: ุงุณู… ุงู„ู…ุณุชุฎุฏู…ุŒ ุชูุถูŠู„ุงุชู‡ุŒ ู…ู‡ุงู…ู‡)
2. ู…ู„ุฎุต ุชู„ู‚ุงุฆูŠ ู„ูƒู„ ุงู„ู…ุญุงุฏุซุงุช ุงู„ุณุงุจู‚ุฉ
3. ุขุฎุฑ N ุฑุณุงู„ุฉ ู„ู„ุณูŠุงู‚ ุงู„ููˆุฑูŠ
4. ู†ุชุงุฆุฌ ุจุญุซ ุฏู„ุงู„ูŠ ุฅุฐุง ูƒุงู† ู‡ู†ุงูƒ ุงุณุชุนู„ุงู…
ู‡ุฐุง ูŠุถู…ู†:
- ุชุฐูƒุฑ ุชุบูŠูŠุฑุงุช ุงู„ู…ูˆุถูˆุน (ูƒู„ ุดูŠุก ู…ูุฏูˆูŽู‘ู†)
- ุงู„ุญูุงุธ ุนู„ู‰ ุงู„ุณูŠุงู‚ ููŠ ุงู„ู…ุญุงุฏุซุงุช ุงู„ุทูˆูŠู„ุฉ ุฌุฏุงู‹
- ุงุณุชุฑุฌุงุน ุงู„ู…ุนู„ูˆู…ุงุช ุฐุงุช ุงู„ุตู„ุฉ ุญุชู‰ ู…ู† ู…ุญุงุฏุซุงุช ู‚ุฏูŠู…ุฉ
"""
if not self.enabled:
return ""
session_id = self._session_id(user_id)
parts = []
try:
# โ”€โ”€โ”€ 1. ุงู„ุฐุงูƒุฑุฉ ุงู„ุฃุณุงุณูŠุฉ (ุญู‚ุงุฆู‚ + ู…ู„ุฎุต + ุฑุณุงุฆู„ ุฃุฎูŠุฑุฉ)
memory = await self.client.memory.get(
session_id=session_id,
min_rating=CONFIG.get("zep_min_fact_rating", 0.3),
)
# ุงู„ุญู‚ุงุฆู‚ ุงู„ู…ุณุชุฎุฑุฌุฉ ุชู„ู‚ุงุฆูŠุงู‹ ู…ู† ูƒู„ ุงู„ู…ุญุงุฏุซุงุช
if hasattr(memory, 'facts') and memory.facts:
facts_lines = []
for f in memory.facts:
fact_text = f if isinstance(f, str) else getattr(f, 'fact', str(f))
if fact_text:
facts_lines.append(f" โ€ข {fact_text}")
if facts_lines:
parts.append(
"### KNOWN FACTS (extracted automatically from all past conversations)\n"
+ "\n".join(facts_lines)
)
# ุงู„ู…ู„ุฎุต ุงู„ุชู„ู‚ุงุฆูŠ ู„ู„ู…ุญุงุฏุซุงุช ุงู„ุณุงุจู‚ุฉ
if hasattr(memory, 'summary') and memory.summary:
summary_text = (
memory.summary if isinstance(memory.summary, str)
else getattr(memory.summary, 'content', '')
)
if summary_text and summary_text.strip():
parts.append(
"### CONVERSATION HISTORY SUMMARY\n"
+ summary_text.strip()
)
# ุขุฎุฑ ุงู„ุฑุณุงุฆู„ ู„ู„ุณูŠุงู‚ ุงู„ููˆุฑูŠ
if hasattr(memory, 'messages') and memory.messages:
n = CONFIG.get("zep_recent_messages", 20)
recent = memory.messages[-n:] if len(memory.messages) > n else memory.messages
if recent:
lines = []
for m in recent:
role = getattr(m, 'role_type', getattr(m, 'role', 'user'))
content = getattr(m, 'content', '') or ''
if content.strip():
lines.append(f" [{role.upper()}]: {content.strip()}")
if lines:
parts.append(
f"### RECENT MESSAGES (last {len(recent)})\n"
+ "\n".join(lines)
)
except Exception as e:
log(f"get_memory error: {e}", "WARN")
# โ”€โ”€โ”€ 2. ุงู„ุจุญุซ ุงู„ุฏู„ุงู„ูŠ ุจู†ุงุกู‹ ุนู„ู‰ ุงู„ุงุณุชุนู„ุงู… ุงู„ุญุงู„ูŠ
if current_query and current_query.strip():
try:
search_results = await self.client.memory.search_sessions(
text=current_query,
user_id=user_id,
limit=CONFIG.get("zep_search_limit", 5),
)
if search_results and hasattr(search_results, 'results') and search_results.results:
relevant_lines = []
for r in search_results.results:
msg = getattr(r, 'message', None)
if msg:
content = getattr(msg, 'content', '') or ''
role = getattr(msg, 'role_type', 'unknown')
score = getattr(r, 'score', 0)
if content.strip() and score > 0.5:
relevant_lines.append(
f" [{role.upper()}] (relevance: {score:.2f}): {content.strip()}"
)
if relevant_lines:
parts.append(
"### SEMANTICALLY RELEVANT PAST CONTEXT (for current query)\n"
+ "\n".join(relevant_lines)
)
except Exception as e:
log(f"semantic search error: {e}", "WARN")
if not parts:
return ""
# โ”€โ”€โ”€ ุจู†ุงุก ุงู„ุณูŠุงู‚ ุงู„ู†ู‡ุงุฆูŠ
context = (
"โ•”โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•—\n"
"โ•‘ PERSISTENT MEMORY โ€” CROSS-SESSION CONTEXT โ•‘\n"
"โ•‘ Use this to maintain full continuity across all chats โ•‘\n"
"โ•šโ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•\n\n"
+ "\n\n".join(parts)
+ "\n\n[END OF PERSISTENT MEMORY]"
)
log(f"Retrieved memory context ({len(context)} chars) for user={user_id}", "MEM")
return context
# โ”€โ”€โ”€ Memory Saving โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
async def save_turn(self, user_id: str, messages: List[dict], assistant_reply: str = ""):
"""
ูŠุญูุธ ุฏูˆุฑุงู† ุงู„ู…ุญุงุฏุซุฉ (ุฑุณุงุฆู„ ุงู„ู…ุณุชุฎุฏู… + ุฑุฏ ุงู„ู…ุณุงุนุฏ) ููŠ Zep.
ูŠูุณุชุฏุนู‰ ุจุนุฏ ูƒู„ ุฑุฏ ู†ุงุฌุญ.
Zep ูŠุณุชุฎุฑุฌ ุงู„ุญู‚ุงุฆู‚ ูˆูŠูุญุฏู‘ุซ ุงู„ู…ู„ุฎุต ุชู„ู‚ุงุฆูŠุงู‹ ููŠ ุงู„ุฎู„ููŠุฉ.
"""
if not self.enabled:
return
session_id = self._session_id(user_id)
zep_msgs = []
try:
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content") or ""
# ุชุฌุงู‡ู„ ุฑุณุงุฆู„ ุงู„ู†ุธุงู… ุฅุฐุง ูƒุงู†ุช ุงู„ุฅุนุฏุงุฏุงุช ุชู…ู†ุนู‡ุง
if role == "system" and not CONFIG.get("zep_save_system_msgs"):
continue
# ุชุญูˆูŠู„ content ุงู„ู‚ุงุฆู…ุฉ ุฅู„ู‰ ู†ุต
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
content = (content or "").strip()
if not content:
continue
# ุชุญูˆูŠู„ tool calls ุฅู„ู‰ ู†ุต ู‚ุงุจู„ ู„ู„ู‚ุฑุงุกุฉ
tool_calls_data = msg.get("tool_calls")
if tool_calls_data and role == "assistant":
tc_summary = json.dumps(
[{"tool": tc["function"]["name"], "args": tc["function"].get("arguments", "{}")}
for tc in tool_calls_data],
ensure_ascii=False
)
content = f"{content}\n[Tool calls: {tc_summary}]" if content else f"[Tool calls: {tc_summary}]"
role_map = {
"user": "user",
"assistant": "assistant",
"tool": "tool",
"system": "system",
}
zep_role = role_map.get(role, "user")
zep_msgs.append(
ZepMessage(
role_type=zep_role,
role=role,
content=content,
)
)
# ุฅุถุงูุฉ ุฑุฏ ุงู„ู…ุณุงุนุฏ ุฅุฐุง ู„ู… ูŠูƒู† ููŠ ุงู„ู‚ุงุฆู…ุฉ
if assistant_reply and assistant_reply.strip():
already_saved = any(
m.get("role") == "assistant" and (m.get("content") or "") == assistant_reply
for m in messages
)
if not already_saved:
zep_msgs.append(
ZepMessage(
role_type="assistant",
role="assistant",
content=assistant_reply.strip(),
)
)
if zep_msgs:
await self.client.memory.add(
session_id=session_id,
messages=zep_msgs,
)
log(f"๐Ÿ’พ Saved {len(zep_msgs)} messages to Zep (user={user_id})", "MEM")
except Exception as e:
log(f"save_turn error: {e}", "WARN")
async def add_fact(self, user_id: str, fact: str):
"""
ุฅุถุงูุฉ ุญู‚ูŠู‚ุฉ ุตุฑูŠุญุฉ ูŠุฐูƒุฑู‡ุง ุงู„ู…ุณุชุฎุฏู… (ู„ู„ุฃุดูŠุงุก ุงู„ู…ู‡ู…ุฉ ุฌุฏุงู‹).
Zep ูŠุณุชุฎุฑุฌ ุงู„ุญู‚ุงุฆู‚ ุชู„ู‚ุงุฆูŠุงู‹ ุฃูŠุถุงู‹ุŒ ู„ูƒู† ู‡ุฐุง ู„ู„ุชุฃูƒูŠุฏ.
"""
if not self.enabled:
return
try:
session_id = self._session_id(user_id)
await self.client.memory.add(
session_id=session_id,
messages=[
ZepMessage(
role_type="system",
role="system",
content=f"[IMPORTANT FACT]: {fact}",
)
]
)
log(f"Added explicit fact for user={user_id}: {fact[:50]}...", "MEM")
except Exception as e:
log(f"add_fact error: {e}", "WARN")
def status(self) -> dict:
return {
"enabled": self.enabled,
"provider": "Zep Cloud",
"error": self._init_error if not self.enabled else None,
}
# ุงู„ู…ุซูŠู„ ุงู„ุนุงู…
zep_memory = ZepMemoryManager()
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Cookie & Headers โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def load_cookie() -> Tuple[str, Optional[str]]:
inline = CONFIG.get("_cookie_inline", "")
if inline:
if inline.startswith("{"):
try:
data = json.loads(inline)
return data.get("cookie", ""), data.get("sapisid") or None
except Exception:
pass
pairs = dict(p.split("=", 1) for p in inline.split("; ") if "=" in p)
return inline, pairs.get("SAPISID") or None
cookie_file = CONFIG.get("cookie_file")
if not cookie_file or not os.path.exists(cookie_file):
return "", None
try:
with open(cookie_file) as f:
content = f.read().strip()
if content.startswith("{"):
data = json.loads(content)
return data.get("cookie", ""), data.get("sapisid") or None
pairs = dict(p.split("=", 1) for p in content.split("; ") if "=" in p)
return content, pairs.get("SAPISID") or None
except Exception as e:
log(f"Cookie load error: {e}", "ERROR")
return "", None
def make_sapisidhash(sapisid: str) -> str:
ts = int(time.time())
h = hashlib.sha1(f"{ts} {sapisid} https://gemini.google.com".encode()).hexdigest()
return f"SAPISIDHASH {ts}_{h}"
def _build_headers() -> dict:
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Origin": "https://gemini.google.com",
"Referer": "https://gemini.google.com/app",
"X-Same-Domain": "1",
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
}
cookie_str, sapisid = load_cookie()
if cookie_str:
headers["Cookie"] = cookie_str
if sapisid:
headers["Authorization"] = make_sapisidhash(sapisid)
return headers
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Gemini Protocol โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def _build_gemini_payload(prompt: str, model_id: int, think_mode: int) -> Tuple[bytes, str]:
inner = [None] * 80
inner[0] = [prompt, 0, None, None, None, None, 0]
inner[1] = ["en"]
inner[2] = ["", "", "", None, None, None, None, None, None, ""]
inner[6] = [0]
inner[7] = 1
inner[10] = 1
inner[11] = 0
inner[17] = [[think_mode]]
inner[18] = 0
inner[27] = 1
inner[30] = [4]
inner[41] = [2]
inner[53] = 0
inner[59] = str(uuid.uuid4())
inner[61] = []
inner[68] = 1
inner[79] = model_id
outer = [None, json.dumps(inner)]
body = urllib.parse.urlencode({"f.req": json.dumps(outer)}).encode()
reqid = int(time.time()) % 1000000
url = (
"https://gemini.google.com/_/BardChatUi/data/"
"assistant.lamda.BardFrontendService/StreamGenerate"
f"?bl={CONFIG['gemini_bl']}&hl=en&_reqid={reqid}&rt=c"
)
return body, url
def gemini_stream_generate(prompt: str, model_id: int, think_mode: int) -> str:
body, url = _build_gemini_payload(prompt, model_id, think_mode)
headers = _build_headers()
proxy = CONFIG.get("proxy")
last_err = None
for attempt in range(CONFIG["retry_attempts"]):
try:
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
ctx = ssl.create_default_context()
if proxy:
opener = urllib.request.build_opener(
urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
urllib.request.HTTPSHandler(context=ctx)
)
resp = opener.open(req, timeout=CONFIG["request_timeout_sec"])
else:
resp = urllib.request.urlopen(req, context=ctx, timeout=CONFIG["request_timeout_sec"])
return resp.read().decode("utf-8", errors="replace")
except Exception as e:
last_err = e
if attempt < CONFIG["retry_attempts"] - 1:
log(f"Retry {attempt+1}/{CONFIG['retry_attempts']}: {e}", "WARN")
time.sleep(CONFIG["retry_delay_sec"])
raise last_err
def clean_gemini_text(text: str) -> str:
text = re.sub(
r'```(?:python|javascript|text)\?code_(?:reference|stdout)&code_event_index=\d+\n.*?```\n?',
'', text, flags=re.DOTALL
)
return text.strip()
def extract_response_text(raw: str) -> str:
texts = []
for line in raw.split("\n"):
if '"wrb.fr"' not in line or len(line) < 200:
continue
try:
arr = json.loads(line)
inner_str = arr[0][2]
if not inner_str or len(inner_str) < 50:
continue
inner = json.loads(inner_str)
if isinstance(inner, list) and len(inner) > 4 and inner[4]:
for part in inner[4]:
if isinstance(part, list) and len(part) > 1 and part[1]:
if isinstance(part[1], list):
for t in part[1]:
if isinstance(t, str) and len(t) > 0:
texts.append(t)
except (json.JSONDecodeError, IndexError, TypeError):
pass
text = ""
for t in reversed(texts):
if t.strip():
text = t
break
return clean_gemini_text(text)
def gemini_stream_iter(prompt: str, model_id: int, think_mode: int):
"""Streaming ุญู‚ูŠู‚ูŠ ูŠูุนูŠุฏ ุงู„ู†ุต ุชุฏุฑูŠุฌูŠุงู‹."""
body_bytes, url = _build_gemini_payload(prompt, model_id, think_mode)
body_str = body_bytes.decode()
headers = _build_headers()
proxy = CONFIG.get("proxy")
if not HAS_HTTPX:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
if text:
yield text
return
prev_text = ""
transport = httpx.HTTPTransport(proxy=proxy) if proxy else None
with httpx.Client(transport=transport, timeout=CONFIG["request_timeout_sec"], verify=True) as client:
with client.stream("POST", url, content=body_str, headers=headers) as resp:
buf = ""
for chunk in resp.iter_text():
buf += chunk
while "\n" in buf:
line, buf = buf.split("\n", 1)
if '"wrb.fr"' not in line or len(line) < 200:
continue
try:
arr = json.loads(line)
inner_str = arr[0][2]
if not inner_str or len(inner_str) < 50:
continue
inner2 = json.loads(inner_str)
if isinstance(inner2, list) and len(inner2) > 4 and inner2[4]:
for part in inner2[4]:
if (isinstance(part, list) and len(part) > 1
and part[1] and isinstance(part[1], list)):
for t in part[1]:
if isinstance(t, str) and len(t) > len(prev_text):
delta = t[len(prev_text):]
delta = clean_gemini_text(delta)
if delta:
yield delta
prev_text = t
except (json.JSONDecodeError, IndexError, TypeError):
pass
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Tool Call Parser โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def parse_tool_calls(text: str) -> Tuple[str, Optional[List[dict]]]:
"""
ู…ุญู„ู‘ู„ Tool Calls ู…ุชุนุฏุฏ ุงู„ู…ุณุชูˆูŠุงุช (4 ุฃู†ู…ุงุท).
"""
tool_calls = []
# ุงู„ู†ู…ุท 1: ```tool_call {...}```
p1 = r'```tool_call\s*\n(.*?)\n?```'
for m in re.finditer(p1, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# ุงู„ู†ู…ุท 2: ```json {...}``` (ูŠุญุชูˆูŠ ุนู„ู‰ name + arguments)
if not tool_calls:
p2 = r'```json\s*\n(.*?)\n?```'
for m in re.finditer(p2, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# ุงู„ู†ู…ุท 3: <tool_call>...</tool_call>
if not tool_calls:
p3 = r'<tool_call>(.*?)</tool_call>'
for m in re.finditer(p3, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# ุงู„ู†ู…ุท 4: JSON ู…ุจุงุดุฑ
if not tool_calls:
jp = r'\{[^{}]*"name"\s*:\s*"[^"]+[^{}]*"arguments"\s*:[^{}]*\}'
for m in re.finditer(jp, text, re.DOTALL):
tc = _try_parse_tc(m.group(0))
if tc:
tool_calls.append(tc)
if not tool_calls:
return text, None
# ุชู†ุธูŠู ุงู„ู†ุต
clean = re.sub(p1, '', text, flags=re.DOTALL)
clean = re.sub(r'```json\s*\n.*?\n?```', '', clean, flags=re.DOTALL)
clean = re.sub(r'<tool_call>.*?</tool_call>', '', clean, flags=re.DOTALL)
return clean.strip(), tool_calls
def _try_parse_tc(raw: str) -> Optional[dict]:
try:
data = json.loads(raw)
except json.JSONDecodeError:
try:
data = json.loads(_fix_json(raw))
except json.JSONDecodeError:
return None
if not isinstance(data, dict):
return None
name = data.get("name") or data.get("function") or data.get("tool")
if not name:
return None
args = data.get("arguments") or data.get("parameters") or data.get("args") or {}
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {"value": args}
return {
"id": f"call_{uuid.uuid4().hex[:8]}",
"type": "function",
"function": {
"name": name,
"arguments": json.dumps(args, ensure_ascii=False),
},
}
def _fix_json(s: str) -> str:
s = re.sub(r',\s*([}\]])', r'\1', s)
s = re.sub(r'(\w+)\s*:', r'"\1":', s)
s = re.sub(r'""(\w+)""', r'"\1"', s)
return s
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Agent System Prompt โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
AGENT_SYSTEM_PROMPT = """You are a professional AI assistant with persistent memory and access to tools.
## MEMORY INSTRUCTIONS
- The [PERSISTENT MEMORY] block above contains everything you remember about this user from ALL previous conversations.
- Use this memory to maintain full continuity even when topics change suddenly.
- If the user references something from a past conversation, look it up in the memory block first.
- Facts, preferences, tasks, and details from previous chats are all available to you.
## TOOL USAGE
When you need to call a tool, use this EXACT format:
```tool_call
{"name": "tool_name_here", "arguments": {"param1": "value1", "param2": "value2"}}
```
### Rules:
1. Always use tools when they are needed โ€” do not guess information a tool can provide.
2. Call tools one at a time and wait for the result before calling the next.
3. Incorporate tool results into your final answer.
4. Never fabricate tool results.
5. If a tool fails, explain what happened and try an alternative approach.
## AVAILABLE TOOLS
{tool_definitions}
## RESPONSE STYLE
- Be thorough and professional in task execution.
- Acknowledge when you remember something from a previous conversation.
- If topics change mid-conversation, handle the new topic fully while retaining all past context.
- Respond in the same language the user uses.
"""
def build_tool_definitions_text(tools: List[dict]) -> str:
if not tools:
return "No tools currently available."
lines = []
for i, tool in enumerate(tools, 1):
fn = tool.get("function", tool) if tool.get("type") == "function" else tool
name = fn.get("name", f"tool_{i}")
desc = fn.get("description", "No description")
params = fn.get("parameters", {})
lines.append(f"### {i}. `{name}`")
lines.append(f"**Description:** {desc}")
if params and params.get("properties"):
lines.append("**Parameters:**")
for pname, pinfo in params["properties"].items():
req = pname in params.get("required", [])
ptype = pinfo.get("type", "any")
pdesc = pinfo.get("description", "")
lines.append(f" - `{pname}` ({ptype}){'*' if req else ''}: {pdesc}")
lines.append("")
return "\n".join(lines)
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Prompt Builder โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def messages_to_prompt(
messages: list,
tools: list = None,
memory_context: str = "",
) -> str:
"""
ูŠุจู†ูŠ prompt ูƒุงู…ู„ ุจุงู„ุชุฑุชูŠุจ:
1. ุฐุงูƒุฑุฉ Zep (ุงู„ุณูŠุงู‚ ุงู„ุฏุงุฆู… ุนุจุฑ ุงู„ุฌู„ุณุงุช)
2. System prompt ุงู„ุฃุฏูˆุงุช (ุฅุฐุง ูƒุงู†ุช ู‡ู†ุงูƒ ุฃุฏูˆุงุช)
3. ุฑุณุงุฆู„ ุงู„ู…ุญุงุฏุซุฉ ุงู„ุญุงู„ูŠุฉ
"""
parts = []
# โ”€โ”€โ”€ 1. ุงู„ุฐุงูƒุฑุฉ ุงู„ุฏุงุฆู…ุฉ ุฃูˆู„ุงู‹ (ุงู„ุฃุนู„ู‰ ุฃูˆู„ูˆูŠุฉ)
if memory_context and memory_context.strip():
parts.append(memory_context)
# โ”€โ”€โ”€ 2. System prompt ู„ู„ุฃุฏูˆุงุช
if tools:
tool_defs = build_tool_definitions_text(tools)
sys_prompt = AGENT_SYSTEM_PROMPT.replace("{tool_definitions}", tool_defs)
parts.append(f"[SYSTEM โ€” AGENT INSTRUCTIONS]\n{sys_prompt}")
else:
parts.append(
"[SYSTEM]\nYou are a professional AI assistant with persistent memory. "
"Use the memory context above to maintain continuity across all conversations."
)
# โ”€โ”€โ”€ 3. ู…ุนุงู„ุฌุฉ ุฑุณุงุฆู„ ุงู„ู…ุญุงุฏุซุฉ
for msg in messages:
role = msg.get("role", "user") if isinstance(msg, dict) else msg.role
content = msg.get("content", "") if isinstance(msg, dict) else (msg.content or "")
tc_data = msg.get("tool_calls") if isinstance(msg, dict) else getattr(msg, "tool_calls", None)
# ุชุญูˆูŠู„ content ู…ุชุนุฏุฏ ุงู„ุฃุฌุฒุงุก
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
content = (content or "").strip()
if role == "system":
# ุฏู…ุฌ system prompt ุงู„ู…ุณุชุฎุฏู… ู…ุน ู…ุง ุณุจู‚
if content:
parts.append(f"[ADDITIONAL SYSTEM INSTRUCTIONS]\n{content}")
elif role == "user":
if content:
parts.append(f"[USER]: {content}")
elif role == "assistant":
if tc_data:
tc_str = "\n".join(
f"```tool_call\n{json.dumps({'name': tc['function']['name'], 'arguments': json.loads(tc['function'].get('arguments', '{}'))}, ensure_ascii=False)}\n```"
for tc in tc_data
)
assistant_part = ""
if content:
assistant_part = f"[ASSISTANT]: {content}\n"
assistant_part += f"[ASSISTANT called tools]:\n{tc_str}"
parts.append(assistant_part)
elif content:
parts.append(f"[ASSISTANT]: {content}")
elif role == "tool":
tool_name = msg.get("name", "") if isinstance(msg, dict) else getattr(msg, "name", "")
tool_id = msg.get("tool_call_id", "") if isinstance(msg, dict) else getattr(msg, "tool_call_id", "")
label = tool_name or tool_id or "unknown_tool"
if content:
parts.append(
f"[TOOL RESULT โ€” {label}]:\n```json\n{content}\n```\n"
f"โ†’ Continue based on this result."
)
return "\n\n".join(p for p in parts if p and p.strip())
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Agent Loop โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def run_agent_sync(
messages: List[dict],
model_id: int,
think_mode: int,
tools: Optional[List[dict]],
memory_context: str = "",
max_iterations: int = None,
) -> Tuple[str, Optional[List[dict]], List[dict]]:
"""
Agent Loop (ReAct):
Reason โ†’ Act (tool call) โ†’ Observe (result) โ†’ Reason โ†’ ...
ูŠุนูŠุฏ (final_text, tool_calls_if_any, updated_messages)
ู…ู„ุงุญุธุฉ: ููŠ ูˆุถุน non-executor (ุงู„ุงูุชุฑุงุถูŠ)ุŒ ูŠูุนูŠุฏ tool calls ู„ู„ู€ client
ู„ูŠูู†ูู‘ุฐู‡ุง ูˆูŠูุฑุณู„ ุงู„ู†ุชุงุฆุฌ ููŠ ุงู„ุทู„ุจ ุงู„ุชุงู„ูŠ. ู‡ุฐุง ู‡ูˆ ุงู„ุณู„ูˆูƒ ุงู„ุตุญูŠุญ
ู„ุจุฑูˆุชูˆูƒูˆู„ OpenAI Tool Calling.
"""
if max_iterations is None:
max_iterations = CONFIG["agent_max_iterations"]
current_messages = list(messages)
for iteration in range(1, max_iterations + 1):
log(f"Agent iteration {iteration}/{max_iterations}", "AGENT")
# ุจู†ุงุก ุงู„ู€ prompt ู…ุน ุงู„ุฐุงูƒุฑุฉ
prompt = messages_to_prompt(current_messages, tools, memory_context)
try:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
except Exception as e:
log(f"Gemini call error (iter {iteration}): {e}", "ERROR")
return f"[Error communicating with model: {e}]", None, current_messages
if not text:
log("Empty response from Gemini", "WARN")
if iteration == 1:
return "[No response from model]", None, current_messages
break
# ุชุญู„ูŠู„ tool calls
clean_text, tool_calls = parse_tool_calls(text)
if tool_calls and tools:
names = [tc["function"]["name"] for tc in tool_calls]
log(f"Tool calls detected: {names}", "TOOL")
# ุฃุถู ุฑุฏ ุงู„ู…ุณุงุนุฏ ู„ู„ุณูŠุงู‚
current_messages.append({
"role": "assistant",
"content": clean_text or None,
"tool_calls": tool_calls,
})
# ุฃุนุฏ tool calls ู„ู„ู€ client (OpenAI protocol)
return clean_text or "", tool_calls, current_messages
else:
# ุฑุฏ ู†ู‡ุงุฆูŠ
log(f"โœ… Final answer at iteration {iteration}", "AGENT")
return clean_text or text, None, current_messages
log(f"Max iterations ({max_iterations}) reached", "WARN")
return text or "[Max iterations reached]", None, current_messages
def run_agent_with_executor_sync(
messages: List[dict],
model_id: int,
think_mode: int,
tools: Optional[List[dict]],
tool_executor: Optional[Callable[[str, dict], str]],
memory_context: str = "",
max_iterations: int = None,
) -> Tuple[str, Optional[List[dict]], List[dict]]:
"""
Agent Loop ุงู„ูƒุงู…ู„ ู…ุน ุชู†ููŠุฐ ุงู„ุฃุฏูˆุงุช ุฏุงุฎู„ูŠุงู‹.
ูŠุณุชุฎุฏู… ุนู†ุฏู…ุง ูŠูƒูˆู† tool_executor ู…ุชุงุญุงู‹ (ู†ุงุฏุฑ ููŠ ู‡ุฐุง ุงู„ุณูŠู†ุงุฑูŠูˆ).
"""
if max_iterations is None:
max_iterations = CONFIG["agent_max_iterations"]
current_messages = list(messages)
for iteration in range(1, max_iterations + 1):
log(f"[Full Agent] Iteration {iteration}/{max_iterations}", "AGENT")
prompt = messages_to_prompt(current_messages, tools, memory_context)
try:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
except Exception as e:
log(f"Gemini error: {e}", "ERROR")
return f"Error: {e}", None, current_messages
if not text:
break
clean_text, tool_calls = parse_tool_calls(text)
if not tool_calls or not tools:
log(f"โœ… Final answer at iteration {iteration}", "AGENT")
return clean_text or text, None, current_messages
current_messages.append({
"role": "assistant",
"content": clean_text or None,
"tool_calls": tool_calls,
})
for tc in tool_calls:
fn_name = tc["function"]["name"]
try:
fn_args = json.loads(tc["function"]["arguments"])
except Exception:
fn_args = {}
log(f"Executing: {fn_name}({fn_args})", "TOOL")
if tool_executor:
try:
result = tool_executor(fn_name, fn_args)
result_str = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
except Exception as e:
result_str = json.dumps({"error": str(e)})
log(f"Tool error: {e}", "ERROR")
else:
result_str = json.dumps({"status": "no_executor", "tool": fn_name})
current_messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"name": fn_name,
"content": result_str,
})
return text or "", None, current_messages
def call_gemini_sync(
prompt: str,
model_id: int,
think_mode: int,
tools: list = None,
) -> Tuple[str, Optional[list]]:
"""ุงุณุชุฏุนุงุก ุจุณูŠุท ุจุฏูˆู† agent loop."""
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
tool_calls = None
if tools and text:
text, tool_calls = parse_tool_calls(text)
return text or "", tool_calls
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Model Resolution โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def resolve_model(model_name: str) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str]]:
think_override = None
if "@think=" in model_name:
model_name, think_str = model_name.rsplit("@think=", 1)
try:
think_override = int(think_str)
except ValueError:
pass
cfg = MODELS.get(model_name)
if not cfg:
default = CONFIG.get("default_model", "gemini-3.5-flash")
cfg = MODELS.get(default)
if cfg:
log(f"Unknown model '{model_name}', falling back to '{default}'", "WARN")
model_name = default
else:
return None, None, None, f"Unknown model: {model_name}. Available: {list(MODELS.keys())}"
return model_name, cfg["mode"], (think_override if think_override is not None else cfg["think"]), None
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Helpers โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def extract_last_user_query(messages: List[dict]) -> str:
"""ูŠุณุชุฎุฑุฌ ุขุฎุฑ ุฑุณุงู„ุฉ ู„ู„ู…ุณุชุฎุฏู… ูƒู†ุต ู„ู„ุจุญุซ ุงู„ุฏู„ุงู„ูŠ."""
for msg in reversed(messages):
if msg.get("role") == "user":
content = msg.get("content", "")
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
return (content or "").strip()
return ""
def _build_completion_response(
cid: str, created: int, model: str, text: str,
tool_calls: Optional[list], prompt: str = ""
) -> dict:
msg: dict = {"role": "assistant", "content": text or None}
if tool_calls:
msg["tool_calls"] = tool_calls
finish = "tool_calls" if tool_calls else "stop"
return {
"id": cid,
"object": "chat.completion",
"created": created,
"model": model,
"choices": [{"index": 0, "message": msg, "finish_reason": finish}],
"usage": {
"prompt_tokens": len(prompt) // 4,
"completion_tokens": len(text) // 4 if text else 0,
"total_tokens": (len(prompt) + len(text)) // 4 if text else len(prompt) // 4,
},
}
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Authentication โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
security = HTTPBearer(auto_error=False)
async def verify_api_key(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
) -> str:
api_keys = CONFIG.get("api_keys", [])
if not api_keys:
return "anonymous"
if not credentials:
raise HTTPException(status_code=401, detail="Missing Authorization header")
token = credentials.credentials
if token not in api_keys:
raise HTTPException(status_code=401, detail="Invalid API key")
return token
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ FastAPI App โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@asynccontextmanager
async def lifespan(app: FastAPI):
await services.init_db()
await services.start_cleanup()
await zep_memory.init() # โ† ุชู‡ูŠุฆุฉ Zep Cloud
log(f"gemini-web2api v{__version__} started", "INFO")
log(f"Memory: {zep_memory.status()}", "MEM")
yield
if services._cleanup_task:
services._cleanup_task.cancel()
app = FastAPI(
title="Gemini Web2API + Zep Memory (Agent Edition)",
version=__version__,
description="OpenAI-compatible Gemini proxy with persistent Zep Cloud memory and full Agent Loop",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ /v1/models โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@app.get("/v1/models")
async def list_models():
return {
"object": "list",
"data": [
{
"id": name,
"object": "model",
"created": 1700000000,
"owned_by": "google",
"description": cfg["desc"],
"capabilities": {
"tool_use": True,
"streaming": True,
"agent_loop": True,
"persistent_memory": zep_memory.enabled,
},
}
for name, cfg in MODELS.items()
],
}
@app.get("/v1beta/models")
async def google_models_list():
return {
"models": [
{
"name": f"models/{name}",
"displayName": name,
"description": cfg["desc"],
"supportedGenerationMethods": ["generateContent", "streamGenerateContent"],
}
for name, cfg in MODELS.items()
]
}
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ /v1/chat/completions โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@app.post("/v1/chat/completions")
async def chat_completions(
body: ChatCompletionRequest,
request: Request,
_: str = Depends(verify_api_key),
):
# โ”€โ”€โ”€ Rate Limiting
client_ip = request.client.host
allowed, wait_time = await services.check_rate_limit(client_ip)
if not allowed:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded. Retry in {wait_time}s.",
headers={"Retry-After": str(wait_time)},
)
# โ”€โ”€โ”€ Model Resolution
model_name, model_id, think_mode, err = resolve_model(
body.model or CONFIG["default_model"]
)
if err:
raise HTTPException(status_code=400, detail=err)
tools = body.tools
messages = [m.dict(exclude_none=True) for m in body.messages]
# user_id: ู†ุณุชุฎุฏู…ู‡ ูƒู…ูุชุงุญ ู„ู„ุฐุงูƒุฑุฉ โ€” ุซุงุจุช ู„ูƒู„ ู…ุณุชุฎุฏู…
user_id = body.user or client_ip or "default"
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โœ… ZEP: ุชุฌู‡ูŠุฒ ุงู„ุฌู„ุณุฉ ูˆุงุณุชุฑุฌุงุน ุงู„ุฐุงูƒุฑุฉ ุงู„ูƒุงู…ู„ุฉ
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
# ุงุณุชุฑุฌุงุน ุงู„ุฐุงูƒุฑุฉ ูˆุงู„ุจุญุซ ุงู„ุฏู„ุงู„ูŠ ุจุงู„ุชูˆุงุฒูŠ
memory_context = await zep_memory.get_full_context(user_id, last_query)
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
cid = f"chatcmpl-{uuid.uuid4().hex[:12]}"
created = int(time.time())
# Cache key (ุจุฏูˆู† ุฐุงูƒุฑุฉ ู„ุฃู† ุงู„ุฐุงูƒุฑุฉ ุชุชุบูŠุฑ)
cache_key = hashlib.md5(
f"{model_name}:{user_id}:{last_query[:200]}:{bool(tools)}".encode()
).hexdigest()
# โ”€โ”€โ”€ Streaming โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
if body.stream:
async def stream_gen():
full_text = ""
final_tool_calls = None
try:
if tools:
loop = asyncio.get_event_loop()
text, tc, updated_msgs = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
full_text = text or ""
final_tool_calls = tc
if tc:
# ุฅุฑุณุงู„ tool calls chunk
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"role": "assistant", "content": full_text or None, "tool_calls": tc},
"finish_reason": "tool_calls",
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
else:
if full_text:
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"role": "assistant", "content": full_text},
"finish_reason": None,
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
else:
# Streaming ุจุฏูˆู† ุฃุฏูˆุงุช
prompt = messages_to_prompt(messages, memory_context=memory_context)
for delta in gemini_stream_iter(prompt, model_id, think_mode):
full_text += delta
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"content": delta},
"finish_reason": None,
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
# Final chunk
finish = "tool_calls" if final_tool_calls else "stop"
yield f"data: {json.dumps({'id': cid, 'object': 'chat.completion.chunk', 'created': created, 'model': model_name, 'choices': [{'index': 0, 'delta': {}, 'finish_reason': finish}]})}\n\n"
yield "data: [DONE]\n\n"
# โœ… ZEP: ุญูุธ ุงู„ู…ุญุงุฏุซุฉ ุจุนุฏ ุงู„ุงู†ุชู‡ุงุก
if full_text and not final_tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, full_text)
except Exception as e:
log(f"Stream error: {e}", "ERROR")
yield f"data: {json.dumps({'error': {'message': str(e), 'type': 'server_error'}})}\n\n"
return StreamingResponse(stream_gen(), media_type="text/event-stream")
# โ”€โ”€โ”€ Non-streaming โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Cache (ุจุฏูˆู† ุฃุฏูˆุงุช ูู‚ุท)
if not tools:
cached = await services.get_cache(cache_key)
if cached:
return JSONResponse(content=json.loads(cached))
try:
loop = asyncio.get_event_loop()
if tools:
text, tool_calls, updated_msgs = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
else:
prompt = messages_to_prompt(messages, memory_context=memory_context)
text, tool_calls = await loop.run_in_executor(
None, call_gemini_sync,
prompt, model_id, think_mode, None
)
except Exception as e:
log(f"Error: {e}", "ERROR")
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
# โœ… ZEP: ุญูุธ ุงู„ู…ุญุงุฏุซุฉ
if text and not tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, text)
prompt_text = messages_to_prompt(messages)
resp_data = _build_completion_response(cid, created, model_name, text, tool_calls, prompt_text)
if not tools and text:
await services.set_cache(cache_key, json.dumps(resp_data))
return JSONResponse(content=resp_data)
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ /v1/responses (Codex CLI / Responses API) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@app.post("/v1/responses")
async def responses_api(
request: Request,
_: str = Depends(verify_api_key),
):
req = await request.json()
model_name, model_id, think_mode, err = resolve_model(
req.get("model", CONFIG["default_model"])
)
if err:
raise HTTPException(status_code=400, detail=err)
input_items = req.get("input", [])
tools = req.get("tools")
messages = []
user_id = req.get("user", "default")
if req.get("instructions"):
messages.append({"role": "system", "content": req["instructions"]})
if isinstance(input_items, str):
messages.append({"role": "user", "content": input_items})
elif isinstance(input_items, list):
for item in input_items:
if isinstance(item, str):
messages.append({"role": "user", "content": item})
elif isinstance(item, dict):
if item.get("type") == "function_call_output":
messages.append({
"role": "tool",
"tool_call_id": item.get("call_id", ""),
"name": item.get("name", ""),
"content": item.get("output", ""),
})
elif item.get("role") == "assistant":
cp = item.get("content", [])
text_acc, tc_list = "", []
if isinstance(cp, list):
for c in cp:
if isinstance(c, dict):
if c.get("type") == "output_text":
text_acc += c.get("text", "")
elif c.get("type") == "function_call":
tc_list.append(c)
elif isinstance(cp, str):
text_acc = cp
m = {"role": "assistant", "content": text_acc or None}
if tc_list:
m["tool_calls"] = [
{
"id": tc.get("call_id", f"call_{i}"),
"type": "function",
"function": {
"name": tc.get("name", ""),
"arguments": tc.get("arguments", "{}"),
}
}
for i, tc in enumerate(tc_list)
]
messages.append(m)
else:
role = item.get("role", "user")
content = item.get("content", "")
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if c.get("type") in ("text", "input_text")
)
messages.append({"role": role, "content": content})
if tools:
tools = [
{
"type": "function",
"function": {
"name": t["name"],
"description": t.get("description", ""),
"parameters": t.get("parameters", {}),
}
}
if t.get("type") == "function" and "function" not in t else t
for t in tools
]
if not any(m.get("content") for m in messages):
raise HTTPException(status_code=400, detail="Empty input")
# โœ… Zep Memory
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
memory_context = await zep_memory.get_full_context(user_id, last_query)
loop = asyncio.get_event_loop()
try:
if tools:
text, tool_calls, _ = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
else:
prompt = messages_to_prompt(messages, memory_context=memory_context)
text, tool_calls = await loop.run_in_executor(
None, call_gemini_sync, prompt, model_id, think_mode, None
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
# โœ… ุญูุธ ููŠ Zep
if text and not tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, text)
rid = f"resp_{uuid.uuid4().hex[:16]}"
mid = f"msg_{uuid.uuid4().hex[:12]}"
output = []
if tool_calls:
for tc in tool_calls:
output.append({
"type": "function_call",
"id": tc["id"],
"call_id": tc["id"],
"name": tc["function"]["name"],
"arguments": tc["function"]["arguments"],
"status": "completed",
})
if text or not tool_calls:
output.append({
"type": "message",
"id": mid,
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": text or "", "annotations": []}],
})
prompt_text = messages_to_prompt(messages)
if req.get("stream"):
async def responses_stream():
yield f"event: response.created\ndata: {json.dumps({'type': 'response.created', 'response': {'id': rid, 'object': 'response', 'status': 'in_progress', 'model': model_name, 'output': []}})}\n\n"
for item in output:
if item["type"] == "function_call":
ev = {
"type": "response.function_call_arguments.done",
"item_id": item["id"], "call_id": item["call_id"],
"name": item["name"], "arguments": item["arguments"],
}
yield f"event: response.function_call_arguments.done\ndata: {json.dumps(ev)}\n\n"
elif item["type"] == "message":
for ci, cp in enumerate(item["content"]):
ev = {
"type": "response.output_text.done",
"item_id": item["id"], "content_index": ci, "text": cp["text"],
}
yield f"event: response.output_text.done\ndata: {json.dumps(ev)}\n\n"
resp_obj = {"id": rid, "object": "response", "status": "completed", "model": model_name, "output": output}
yield f"event: response.completed\ndata: {json.dumps({'type': 'response.completed', 'response': resp_obj})}\n\n"
return StreamingResponse(responses_stream(), media_type="text/event-stream")
return JSONResponse({
"id": rid, "object": "response", "created_at": int(time.time()),
"status": "completed", "model": model_name, "output": output,
"usage": {
"input_tokens": len(prompt_text) // 4,
"output_tokens": len(text) // 4 if text else 0,
"total_tokens": (len(prompt_text) + len(text)) // 4 if text else len(prompt_text) // 4,
},
})
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Google Native API /v1beta โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def _google_contents_to_messages(req: dict) -> list:
messages = []
sys_inst = req.get("systemInstruction")
if sys_inst:
sys_text = " ".join(p.get("text", "") for p in sys_inst.get("parts", []) if p.get("text"))
if sys_text:
messages.append({"role": "system", "content": sys_text})
for content in req.get("contents", []):
role = "assistant" if content.get("role") == "model" else "user"
text = " ".join(p.get("text", "") for p in content.get("parts", []) if p.get("text"))
messages.append({"role": role, "content": text})
return messages
@app.post("/v1beta/models/{model_path:path}")
async def google_generate(model_path: str, request: Request):
req = await request.json()
model_name_raw = model_path.split(":")[0]
is_stream = ":streamGenerateContent" in model_path
model_name, model_id, think_mode, err = resolve_model(model_name_raw)
if err:
raise HTTPException(status_code=400, detail=err)
messages = _google_contents_to_messages(req)
if not messages:
raise HTTPException(status_code=400, detail="Empty content")
# Zep ู„ู„ู€ Google API ุฃูŠุถุงู‹
user_id = req.get("user", "google_api")
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
memory_context = await zep_memory.get_full_context(user_id, last_query)
prompt = messages_to_prompt(messages, memory_context=memory_context)
loop = asyncio.get_event_loop()
try:
text, _ = await loop.run_in_executor(
None, call_gemini_sync, prompt, model_id, think_mode, None
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
if text:
await zep_memory.save_turn(user_id, messages, text)
candidate = {
"content": {"parts": [{"text": text or ""}], "role": "model"},
"finishReason": "STOP",
"index": 0,
}
response_obj = {
"candidates": [candidate],
"usageMetadata": {
"promptTokenCount": len(prompt) // 4,
"candidatesTokenCount": len(text) // 4 if text else 0,
"totalTokenCount": (len(prompt) + len(text)) // 4 if text else len(prompt) // 4,
},
"modelVersion": model_name,
}
if is_stream:
async def google_stream():
yield f"data: {json.dumps(response_obj)}\n\n"
return StreamingResponse(google_stream(), media_type="text/event-stream")
return JSONResponse(response_obj)
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Health & Root โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
@app.get("/")
async def root():
return {
"service": "gemini-web2api",
"version": __version__,
"models": list(MODELS.keys()),
"memory": zep_memory.status(),
"features": {
"persistent_memory": zep_memory.enabled,
"agent_loop": True,
"tool_calling": True,
"streaming": True,
"cache": True,
"rate_limiting": True,
},
"endpoints": [
"/v1/chat/completions",
"/v1/models",
"/v1/responses",
"/v1beta/models",
"/health",
"/memory/status",
],
}
@app.get("/health")
async def health():
return {
"status": "healthy",
"timestamp": time.time(),
"version": __version__,
"memory_enabled": zep_memory.enabled,
}
@app.get("/memory/status")
async def memory_status():
"""ุญุงู„ุฉ ุงู„ุฐุงูƒุฑุฉ ุงู„ุฏุงุฆู…ุฉ."""
return {
"provider": "Zep Cloud",
"enabled": zep_memory.enabled,
"config": {
"memory_type": CONFIG.get("zep_memory_type"),
"recent_messages": CONFIG.get("zep_recent_messages"),
"search_limit": CONFIG.get("zep_search_limit"),
"min_fact_rating": CONFIG.get("zep_min_fact_rating"),
},
"error": zep_memory._init_error if not zep_memory.enabled else None,
}
@app.post("/memory/add-fact")
async def add_fact_endpoint(
request: Request,
_: str = Depends(verify_api_key),
):
"""ุฅุถุงูุฉ ุญู‚ูŠู‚ุฉ ุตุฑูŠุญุฉ ู„ุฐุงูƒุฑุฉ ู…ุณุชุฎุฏู… ู…ุญุฏุฏ."""
body = await request.json()
user_id = body.get("user_id", "default")
fact = body.get("fact", "")
if not fact:
raise HTTPException(status_code=400, detail="fact is required")
await zep_memory.ensure_user_session(user_id)
await zep_memory.add_fact(user_id, fact)
return {"status": "ok", "fact": fact, "user_id": user_id}
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Gradio UI โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
def build_gradio_interface():
async def chat_handler(
message: str,
history: list,
model_choice: str,
user_id_box: str,
request: gr.Request,
):
if not message or not message.strip():
yield history
return
model_name, model_id, think_mode, err = resolve_model(model_choice)
if err:
history.append((message, f"โŒ {err}"))
yield history
return
uid = (user_id_box or "").strip() or (request.client.host if request.client else "default")
# ุชุฌู‡ูŠุฒ ุงู„ุฌู„ุณุฉ ูˆุงุณุชุฑุฌุงุน ุงู„ุฐุงูƒุฑุฉ
await zep_memory.ensure_user_session(uid)
memory_context = await zep_memory.get_full_context(uid, message)
history.append((message, "โณ ..."))
yield history
full_response = ""
try:
msgs = [{"role": "user", "content": message}]
prompt = messages_to_prompt(msgs, memory_context=memory_context)
for chunk in gemini_stream_iter(prompt, model_id, think_mode):
full_response += chunk
history[-1] = (message, full_response)
yield history
except Exception as e:
full_response = f"โŒ Error: {e}"
history[-1] = (message, full_response)
yield history
return
# ุญูุธ ููŠ Zep
if full_response:
await zep_memory.save_turn(
uid,
[{"role": "user", "content": message}],
full_response,
)
log(f"Saved chat turn to Zep for user={uid}", "MEM")
mem_status = "โœ… Zep Cloud ู…ุชุตู„" if HAS_ZEP else "โŒ zep-cloud ุบูŠุฑ ู…ุซุจู‘ุช"
api_key_display = CONFIG.get("api_keys", ["sk-gemini"])[0]
with gr.Blocks(
title="Gemini Agent + Zep Memory",
theme=gr.themes.Soft(),
) as demo:
gr.Markdown(f"""
# ๐Ÿค– Gemini Professional Agent + ๐Ÿง  Zep Persistent Memory
### v{__version__} โ€” ุฐุงูƒุฑุฉ ุฏุงุฆู…ุฉ ุนุจุฑ ูƒู„ ุงู„ุฌู„ุณุงุช
| ุงู„ุฎุงุตูŠุฉ | ุงู„ู‚ูŠู…ุฉ |
|---------|--------|
| **Memory** | {mem_status} |
| **Agent Loop** | โœ… ุญุชู‰ {CONFIG['agent_max_iterations']} iterations |
| **API Key** | `{api_key_display}` |
| **Base URL** | `/v1` |
""")
with gr.Row():
with gr.Column(scale=4):
chatbot = gr.Chatbot(
height=520,
label="ุงู„ู…ุญุงุฏุซุฉ โ€” ุชูุญูุธ ุชู„ู‚ุงุฆูŠุงู‹ ููŠ Zep",
show_copy_button=True,
)
with gr.Column(scale=1):
gr.Markdown("""
### ๐Ÿง  ุงู„ุฐุงูƒุฑุฉ ุงู„ุฏุงุฆู…ุฉ
ูŠุชุฐูƒุฑ ุงู„ูˆูƒูŠู„ ุนุจุฑ **ูƒู„ ุงู„ุฌู„ุณุงุช**:
- ุงู„ุญู‚ุงุฆู‚ ุงู„ู…ุณุชุฎุฑุฌุฉ ุชู„ู‚ุงุฆูŠุงู‹
- ู…ู„ุฎุต ูƒุงู…ู„ ู„ู„ู…ุญุงุฏุซุงุช
- ุขุฎุฑ 20 ุฑุณุงู„ุฉ ุจุงู„ุชูุตูŠู„
- ุจุญุซ ุฏู„ุงู„ูŠ ููŠ ูƒู„ ุงู„ุชุงุฑูŠุฎ
### โœ… ุงู„ู…ู…ูŠุฒุงุช
- ูŠุชุฐูƒุฑ ุชุบูŠูŠุฑุงุช ุงู„ู…ูˆุถูˆุน
- ูŠุญูุธ ุชูุงุตูŠู„ ุงู„ู…ู‡ุงู…
- Agent Loop ุงุญุชุฑุงููŠ
- Tool Calling ู…ูˆุซูˆู‚
- Streaming ุญู‚ูŠู‚ูŠ
""")
with gr.Row():
user_id_box = gr.Textbox(
label="๐Ÿ†” User ID (ุงุฎุชูŠุงุฑูŠ โ€” ู„ู„ุชู…ูŠูŠุฒ ุจูŠู† ุงู„ู…ุณุชุฎุฏู…ูŠู†)",
placeholder="ู…ุซุงู„: user_123 ุฃูˆ ุงุชุฑูƒู‡ ูุงุฑุบุงู‹",
scale=2,
)
model_dd = gr.Dropdown(
choices=list(MODELS.keys()),
value=CONFIG["default_model"],
label="๐Ÿค– ุงู„ู†ู…ูˆุฐุฌ",
scale=2,
)
with gr.Row():
msg_box = gr.Textbox(
label="ุฑุณุงู„ุชูƒ",
placeholder="ุงูƒุชุจ ู‡ู†ุง... ูŠู…ูƒู†ูƒ ุชุบูŠูŠุฑ ุงู„ู…ูˆุถูˆุน ููŠ ุฃูŠ ูˆู‚ุชุŒ ุงู„ูˆูƒูŠู„ ูŠุชุฐูƒุฑ ูƒู„ ุดูŠุก โœจ",
lines=3,
scale=5,
)
send_btn = gr.Button("ุฅุฑุณุงู„ โ–ถ", variant="primary", scale=1)
with gr.Row():
clear_btn = gr.Button("๐Ÿ—‘๏ธ ู…ุณุญ ุงู„ุนุฑุถ (ู„ุง ูŠู…ุณุญ ุงู„ุฐุงูƒุฑุฉ)")
# ุฑุจุท ุงู„ุฃุญุฏุงุซ
msg_box.submit(
chat_handler,
[msg_box, chatbot, model_dd, user_id_box],
[chatbot],
).then(lambda: "", None, [msg_box])
send_btn.click(
chat_handler,
[msg_box, chatbot, model_dd, user_id_box],
[chatbot],
).then(lambda: "", None, [msg_box])
clear_btn.click(lambda: [], None, [chatbot])
return demo
demo = build_gradio_interface()
app = gr.mount_gradio_app(app, demo, path="/chat")
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
# โ”€โ”€โ”€ Entry Point โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•
if __name__ == "__main__":
import uvicorn
parser = argparse.ArgumentParser(description="Gemini Web2API + Zep Memory Agent")
parser.add_argument("--port", type=int, default=CONFIG["port"])
parser.add_argument("--host", type=str, default=CONFIG["host"])
parser.add_argument("--cookie-file", type=str, default=None)
parser.add_argument("--proxy", type=str, default=None)
parser.add_argument("--max-iterations", type=int, default=CONFIG["agent_max_iterations"])
parser.add_argument("--zep-key", type=str, default=None, help="Zep Cloud API Key")
parser.add_argument("--version", action="version", version=f"v{__version__}")
args = parser.parse_args()
if args.cookie_file:
CONFIG["cookie_file"] = args.cookie_file
if args.proxy:
CONFIG["proxy"] = args.proxy
if args.zep_key:
CONFIG["zep_api_key"] = args.zep_key
CONFIG["agent_max_iterations"] = args.max_iterations
print(f"\n{'โ•'*65}")
print(f" gemini-web2api v{__version__} โ€” Professional Agent + Zep Memory")
print(f"{'โ•'*65}")
print(f" API: http://localhost:{args.port}/v1")
print(f" Chat UI: http://localhost:{args.port}/chat")
print(f" Health: http://localhost:{args.port}/health")
print(f" Memory: http://localhost:{args.port}/memory/status")
print(f" Models: {', '.join(MODELS.keys())}")
print(f" Agent Loop: max {CONFIG['agent_max_iterations']} iterations")
print(f" Zep Memory: {'โœ… Key configured' if CONFIG.get('zep_api_key') else 'โŒ ZEP_API_KEY not set'}")
print(f" Cookie: {'โœ…' if CONFIG.get('cookie_file') or CONFIG.get('_cookie_inline') else 'โŒ none (anonymous)'}")
print(f" Proxy: {CONFIG.get('proxy') or 'โŒ none'}")
print(f"{'โ•'*65}\n")
uvicorn.run(
app,
host=args.host,
port=args.port,
loop="asyncio",
timeout_keep_alive=120,
)