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 asbobi + sxema tavsifi
├── history.py       ← suhbat tarixini boshqarish
├── agent.py         ← asosiy halqa (+ zaxira model)
└── handlers.py      ← faqat adminlar uchun kirish

-- 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;

# bot/agent/tools.py
import psycopg

RO_DSN = "postgresql://agent_ro:...@127.0.0.1/humo"  # .env'dan olinadi

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)
"""


def run_sql(query: str) -> str:
    """Faqat SELECT; natija matn jadval ko'rinishida (model o'qishi uchun)."""
    lowered = query.strip().lower()
    # read-only rol asosiy himoya; bu tekshiruv — tezkor va tushunarli xato uchun
    if not lowered.startswith("select"):
        return "XATO: faqat SELECT so'rovlarga ruxsat berilgan."
    with psycopg.connect(RO_DSN) as conn:
        conn.execute(f"SET statement_timeout = {QUERY_TIMEOUT_MS}")
        rows = conn.execute(query).fetchmany(MAX_ROWS + 1)
        if not rows:
            return "(bo'sh natija)"
        clipped = rows[:MAX_ROWS]
        suffix = f"\n... (faqat birinchi {MAX_ROWS} qator)" if len(rows) > MAX_ROWS else ""
        return "\n".join(" | ".join(str(v) for v in row) for row in clipped) + suffix

# bot/agent/history.py
HISTORY_MAX = 20


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

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY muhitdan

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


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 = 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 asbob so'radi — bajaramiz va natijani qaytaramiz
        results = [
            {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": run_sql(block.input["query"]),
            }
            for block in response.content
            if block.type == "tool_use"
        ]
        messages.append({"role": "user", "content": results})
    return "So'rov juda murakkab bo'ldi — savolni bo'lib bering.", messages

# bot/agent/handlers.py (asosiy qismi)
@router.message(AdminFilter())
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 = ask(history)
    except anthropic.APIError as exc:
        # JIMGINA boshqa modelga TUSHMAYMIZ. Admin xatoni ko'radi va
        # o'zi qaror qiladi — chunki zaxira model kuchsizroq bo'lishi mumkin
        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 message.answer(answer)

# bot/agent/handlers.py
from aiogram.filters import BaseFilter

ADMIN_IDS = {123456789, 987654321}  # .env'dan


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

# 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()]) == []