AI agent botning ichida: bazaga erkin savol, tool-use va zaxira model

Admin savoli
    │
    ▼
LLM ──"menga SQL kerak"──► run_sql(READ-ONLY!) ──natija──► LLM ──► javob
 ▲                                                          │
 └────────── kerak bo'lsa yana so'rov (halqa) ◄─────────────┘

bot/agent/
├── __init__.py
├── tools.py         ← read-only SQL tool'i + sxema tavsifi
├── history.py       ← suhbat tarixini boshqarish
├── agent.py         ← asosiy halqa
├── fallback.py      ← zaxira model (faqat admin so'rasa)
└── handlers.py      ← faqat adminlar uchun kirish

uv add anthropic "psycopg[binary]"

-- bazada bir marta: faqat SELECT huquqli rol
CREATE USER agent_ro WITH PASSWORD '...';
GRANT CONNECT ON DATABASE humo TO agent_ro;
GRANT USAGE ON SCHEMA public TO agent_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_ro;
-- kelajakdagi jadvallar uchun ham:
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO agent_ro;
-- rolning har tranzaksiyasi read-only boshlansin: vaqtinchalik jadval ham yaratilmaydi
ALTER ROLE agent_ro SET default_transaction_read_only = on;

# bot/agent/tools.py
import os

import psycopg

# faqat SELECT huquqli rol bilan ulanish, .env'da:
# AGENT_RO_DSN=postgresql://agent_ro:...@127.0.0.1/humo
RO_DSN = os.environ.get("AGENT_RO_DSN", "")

MAX_ROWS = 50  # model 10 ming qatorni baribir o'qiy olmaydi
QUERY_TIMEOUT_MS = 5000

# Modelga beriladigan sxema tavsifi — qo'lda yozilgan, muhim jadvallar bilan.
# information_schema'ni to'liq berish shart emas: ortiqcha jadval — ortiqcha adashuv
SCHEMA_DOC = """
Jadvallar:
- orders(id, client_id, direction, weight_kg, volume_m3, price_usd, created_at)
- clients(id, name, phone, created_at)
- calc_requests(id, chat_id, product, result_price, created_at)
"""


async def run_sql(query: str) -> tuple[str, bool]:
    """(natija, xato_bo'ldimi). Natija — matn jadval, model o'qishi uchun."""
    lowered = query.strip().lower()
    # read-only rol asosiy himoya; bu tekshiruv — modelga tez va tushunarli xato uchun
    if not lowered.startswith(("select", "with")):
        return "XATO: faqat SELECT (yoki WITH ... SELECT) so'rovlarga ruxsat berilgan.", True
    try:
        async with await psycopg.AsyncConnection.connect(RO_DSN) as conn:
            # tranzaksiya READ ONLY: rolga tasodifan yozish huquqi berilib qolsa
            # ham, bu ulanish orqali hech narsa yozilmaydi
            await conn.set_read_only(True)
            await conn.execute(f"SET statement_timeout = {QUERY_TIMEOUT_MS}")
            # prepare=True — so'rov faqat BITTA buyruq bo'la oladi:
            # "SELECT 1; DELETE ..." kabi zanjir bazaga umuman yetib bormaydi
            cur = await conn.execute(query, prepare=True)
            rows = await cur.fetchmany(MAX_ROWS + 1)
    except psycopg.Error as exc:
        # xatoni modelga qaytaramiz: ustun nomida adashgan bo'lsa, o'zi tuzatadi
        return f"XATO: {exc}", True
    if not rows:
        return "(bo'sh natija)", False
    clipped = rows[:MAX_ROWS]
    suffix = f"\n... (faqat birinchi {MAX_ROWS} qator)" if len(rows) > MAX_ROWS else ""
    table = "\n".join(" | ".join(str(v) for v in row) for row in clipped)
    return table + suffix, False

# bot/agent/history.py
HISTORY_MAX = 20

# Suhbatlar jarayon xotirasida: restart'da tozalanadi, admin suhbati uchun yetarli.
# Redis yoki bazaga saqlasangiz, javobdagi SDK obyektlarini JSON'ga aylantirish kerak
_histories: dict[int, list] = {}


def load_history(chat_id: int) -> list:
    return list(_histories.get(chat_id, []))


def save_history(chat_id: int, messages: list) -> None:
    _histories[chat_id] = messages


def trim(messages: list[dict]) -> list[dict]:
    """Oxirgi HISTORY_MAX xabar, lekin tarix TOZA user xabaridan boshlanishi shart.

    Yetim tool_result (juftining tool_use'i kesilib ketgan) yoki yolg'iz
    assistant xabari bosh bo'lib qolsa — API 400 beradi. Shuning uchun
    boshlanishni birinchi "toza" user turn'gacha suramiz.
    """
    tail = messages[-HISTORY_MAX:]
    for i, msg in enumerate(tail):
        if msg["role"] != "user":
            continue
        content = msg.get("content")
        # user turn ichida tool_result bo'lsa — bu "javob turn", boshlanish emas
        if isinstance(content, list) and any(
            block.get("type") == "tool_result" for block in content
        ):
            continue
        return tail[i:]
    return []  # toza boshlanish topilmadi — yangi suhbat boshlaganimiz ma'qul

# bot/agent/agent.py
import anthropic

from .tools import SCHEMA_DOC, run_sql

# ANTHROPIC_API_KEY muhitdan. async mijoz: agent bir savolga bir necha marta
# model va bazaga murojaat qiladi — sinxron chaqiruv shu vaqtda butun botni
# to'xtatib qo'yardi. SDK standart holatda 600 soniya kutadi — qisqartiramiz
client = anthropic.AsyncAnthropic(timeout=60, max_retries=1)

SYSTEM = f"""Sen Humo Logistics'ning tahlilchi-agentisan. Savollarga faqat
bazadagi ma'lumot asosida javob ber; bilmasang — bilmayman de, o'ylab topma.
Javob tili — o'zbekcha, qisqa va raqamli.
{SCHEMA_DOC}"""

TOOLS = [
    {
        "name": "run_sql",
        "description": "PostgreSQL bazasida read-only SELECT bajaradi",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    }
]

MAX_TURNS = 6  # cheksiz halqadan saqlovchi: 6 tool-chaqiruvdan keyin to'xtatamiz


async def ask(history: list[dict]) -> tuple[str, list[dict]]:
    """(javob_matni, yangilangan_tarix). Xatoni YUTMAYDI — yuqoriga otadi."""
    messages = list(history)
    for _ in range(MAX_TURNS):
        response = await client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1500,
            system=SYSTEM,
            tools=TOOLS,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            text = "".join(b.text for b in response.content if b.type == "text")
            return text, messages

        # model tool so'radi — bajaramiz va natijani qaytaramiz
        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            output, is_error = await run_sql(block.input["query"])
            results.append(
                {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                    # xato belgisi: model natija emas, xato olganini aniq biladi
                    "is_error": is_error,
                }
            )
        messages.append({"role": "user", "content": results})
    return "So'rov juda murakkab bo'ldi — savolni bo'lib bering.", messages

# bot/agent/fallback.py
import json
from functools import cache

from openai import AsyncOpenAI

from .agent import MAX_TURNS, SYSTEM
from .tools import run_sql

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "run_sql",
            "description": "PostgreSQL bazasida read-only SELECT bajaradi",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    }
]


@cache
def _client() -> AsyncOpenAI:
    return AsyncOpenAI(timeout=60, max_retries=1)  # OPENAI_API_KEY muhitdan


async def ask_fallback(question: str) -> str:
    """Zaxira: tarixsiz, faqat shu savol bilan — o'sha read-only run_sql tool'i."""
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": question},
    ]
    for _ in range(MAX_TURNS):
        response = await _client().chat.completions.create(
            model="gpt-4o", messages=messages, tools=TOOLS
        )
        reply = response.choices[0].message
        if not reply.tool_calls:
            return reply.content or ""
        messages.append(reply)
        for call in reply.tool_calls:
            output, _ = await run_sql(json.loads(call.function.arguments)["query"])
            messages.append({"role": "tool", "tool_call_id": call.id, "content": output})
    return "So'rov juda murakkab bo'ldi — savolni bo'lib bering."

# bot/agent/handlers.py
import logging

from aiogram import F, Router, types
from aiogram.filters import BaseFilter, Command

from bot.agent.agent import ask
from bot.agent.fallback import ask_fallback
from bot.agent.history import load_history, save_history, trim
from bot.reports.notify import ADMIN_IDS  # 3-qism: .env'dagi ro'yxat

router = Router()
log = logging.getLogger(__name__)

TELEGRAM_LIMIT = 4096  # bitta xabar matnining chegarasi
_pending: dict[int, str] = {}  # /fallback'ni kutayotgan savollar


class AdminFilter(BaseFilter):
    async def __call__(self, message: types.Message) -> bool:
        return message.from_user is not None and message.from_user.id in ADMIN_IDS


async def answer_long(message: types.Message, text: str) -> None:
    # 1500 tokenlik javob 4096 belgidan oshishi mumkin — bo'laklab yuboramiz
    text = text or "(bo'sh javob)"
    for start in range(0, len(text), TELEGRAM_LIMIT):
        await message.answer(text[start : start + TELEGRAM_LIMIT])


@router.message(AdminFilter(), F.text, ~F.text.startswith("/"))
async def agent_message(message: types.Message):
    history = trim(load_history(message.chat.id))
    history.append({"role": "user", "content": message.text})
    try:
        answer, new_history = await ask(history)
    except Exception as exc:
        # JIMGINA boshqa modelga TUSHMAYMIZ. Admin xatoni ko'radi va
        # o'zi qaror qiladi — chunki zaxira model kuchsizroq bo'lishi mumkin
        log.exception("agent xatosi")
        _pending[message.chat.id] = message.text
        await message.answer(
            f"Asosiy model xatosi: {type(exc).__name__}.\n"
            "Zaxira (gpt-4o) bilan davom etaymi? /fallback",
        )
        return
    save_history(message.chat.id, new_history)
    await answer_long(message, answer)


@router.message(AdminFilter(), Command("fallback"))
async def agent_fallback(message: types.Message):
    question = _pending.pop(message.chat.id, None)
    if question is None:
        await message.answer("Zaxiraga o'tkaziladigan savol yo'q.")
        return
    try:
        answer = await ask_fallback(question)
    except Exception as exc:
        log.exception("zaxira model xatosi")
        await message.answer(f"Zaxira model ham ishlamadi: {type(exc).__name__}.")
        return
    # javob qaysi modeldan kelgani ko'rinib tursin
    await answer_long(message, f"[zaxira: gpt-4o]\n{answer}")

# bot/app.py — router'lar tartibi
dp.include_router(nav_reset.router)  # 1) avval qutqaruvchi
dp.include_router(login.router)  # 2) keyin dialoglar
dp.include_router(agent.router)  # 3) admin matni — agentga
dp.include_router(ai_calc.router)  # 4) mijoz matni — AI kalkulyatorga, eng oxirida

# tests/test_history.py
from bot.agent.history import trim


def user(text="salom"):
    return {"role": "user", "content": text}


def tool_result_turn():
    return {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1"}]}


def assistant():
    return {"role": "assistant", "content": [{"type": "text", "text": "javob"}]}


def test_orphan_tool_result_dropped_from_front():
    # kesishdan keyin bosh: [tool_result, assistant, user...] — yetimlar tashlanadi
    messages = [tool_result_turn(), assistant(), user("keyingi savol"), assistant()]
    trimmed = trim(messages)
    assert trimmed[0] == user("keyingi savol")


def test_clean_history_untouched():
    messages = [user(), assistant()]
    assert trim(messages) == messages


def test_no_clean_start_resets():
    assert trim([assistant(), tool_result_turn()]) == []

uv run python -m pytest tests -q