Помогите разобраться в проблеме с баном в deepseek, при использовании ds2api, пожалуйста.

При внедрении ds2api в свой проект, я столкнулся с проблемой банов аккаунтов deepseek, используемых, для сгенерированного API.Я даже поставил задержку между запросами, но это не помогло. С каждым разом, продолжительность бана увеличивается. Исходный код проекта:import os

import json

import asyncio

import logging

import random

from pathlib import Path

from typing import Dict, List, Any, Optional

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

from aiogram import Bot, Dispatcher, types

from aiogram.exceptions import TelegramBadRequest

from openai import AsyncOpenAI

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")

logger = logging.getLogger("yasteria.backend")

# Настройки окружения и песочницы

SANDBOX_DIR = Path(os.getenv("SANDBOX_DIR", "ai_workspace")).resolve()

SANDBOX_DIR.mkdir(exist_ok=True)

DS_BASE_URL = os.getenv("DS_BASE_URL", "http://localhost:5001/v1")

DS_API_KEY = os.getenv("DS_API_KEY", "YOUR_DS_API_KEY")

MODEL_NAME = os.getenv("MODEL_NAME", "deepseek-v4-flash")

if DS_API_KEY == "YOUR_DS_API_KEY":

logger.warning("!!! Ключ DS_API_KEY не задан. Запросы к нейронке упадут.")

client = AsyncOpenAI(base_url=DS_BASE_URL, api_key=DS_API_KEY)

# Ограничитель, чтобы не ловить 429 от API

api_limiter = asyncio.Semaphore(1)

BASE_API_DELAY = float(os.getenv("BASE_API_DELAY", "4.0"))

TELEGRAM_TOKENS = {

"Planner": os.getenv("PLANNER_BOT_TOKEN", "YOUR_PLANNER_BOT_TOKEN"),

"Coder": os.getenv("CODER_BOT_TOKEN", "YOUR_CODER_BOT_TOKEN"),

"Explorer": os.getenv("EXPLORER_BOT_TOKEN", "YOUR_EXPLORER_BOT_TOKEN")

}

# Хранилище тасок для управления ботами в фоне

bot_tasks: Dict[str, asyncio.Task] = {}

u/asynccontextmanager

async def lifespan(app: FastAPI):

yield

# При выключении сервера гасим пуллинг всех ботов

if bot_tasks:

logger.info(f"Завершение работы: останавливаем ботов ({len(bot_tasks)} шт.)")

for token, task in bot_tasks.items():

task.cancel()

await asyncio.gather(*bot_tasks.values(), return_exceptions=True)

bot_tasks.clear()

app = FastAPI(title="Yasteria Agents API", lifespan=lifespan)

class AgentQuery(BaseModel):

prompt: str

class BotConfig(BaseModel):

name: str

role: str

token: Optional[str] = None

def _get_safe_path(user_path: str) -> Optional[Path]:

# Защита от .. и выхода за пределы рабочей папки (path traversal)

clean_path = user_path.replace("\\", "/").lstrip("/")

# Срезаем дубли папок, если агент накосячил в путях

prefix = f"{SANDBOX_DIR.name}/"

if clean_path.startswith(prefix):

clean_path = clean_path[len(prefix):]

elif clean_path == SANDBOX_DIR.name:

return None

try:

target_path = (SANDBOX_DIR / clean_path).resolve()

if target_path.is_relative_to(SANDBOX_DIR):

return target_path

except Exception:

pass

return None

def write_file(arguments: Dict[str, Any]) -> str:

path_str = arguments.get("path")

content = arguments.get("content", "")

if not path_str:

return "Ошибка: пустой путь 'path'."

safe_path = _get_safe_path(path_str)

if not safe_path:

logger.warning(f"Попытка выйти из песочницы на запись: {path_str}")

return "Ошибка безопасности: Выход за пределы рабочей директории запрещен."

try:

safe_path.parent.mkdir(parents=True, exist_ok=True)

safe_path.write_text(content, encoding="utf-8")

logger.info(f"Файл записан: {safe_path}")

return f"Успех: Файл записан по пути '{safe_path.relative_to(SANDBOX_DIR)}'."

except Exception as e:

logger.error(f"Сбой записи файла {path_str}: {e}")

return f"Ошибка при записи файла: {e}"

def read_file(arguments: Dict[str, Any]) -> str:

path_str = arguments.get("path")

if not path_str:

return "Ошибка: пустой путь 'path'."

safe_path = _get_safe_path(path_str)

if not safe_path:

logger.warning(f"Попытка выйти из песочницы на чтение: {path_str}")

return "Ошибка безопасности: Доступ к этому файлу запрещен."

try:

if not safe_path.exists():

return "Ошибка: Файл не найден."

content = safe_path.read_text(encoding="utf-8")

return f"Содержимое файла:\n--- START ---\n{content}\n--- END ---"

except Exception as e:

logger.error(f"Сбой чтения файла {path_str}: {e}")

return f"Ошибка при чтении файла: {e}"

def execute_skill(skill_name: str, arguments: Dict[str, Any]) -> str:

# Роутер системных вызовов для Кодера

handlers = {

"write_file": write_file,

"read_file": read_file

}

handler = handlers.get(skill_name)

if handler:

return handler(arguments)

return f"Ошибка: Неизвестный навык '{skill_name}'."

# TODO: Вынести системные промпты в файлы, JSON тут глаза мозолит

AGENTS_CONFIG = {

"Planner": {

"prompt": (

"Ты — Планёр, первый агент в цепочке оркестрации Yasteria.\n"

"Твоя задача — декомпозировать запрос пользователя в техническую задачу для Кодера.\n"

"Проанализируй, какие файлы нужно создать, изменить или прочитать.\n"

"ОТВЕЧАЙ СТРОГО В ФОРМАТЕ JSON:\n"

"{\n"

' "task_description": "ТЗ для Кодера с точными путями и именами файлов",\n'

' "target_directory": "ai_workspace"\n'

"}"

)

},

"Coder": {

"prompt": (

"Ты — Кодер, второй агент в цепочке. Выполняй ТЗ Планёра.\n"

"Используй физические навыки сервера (Skills) через JSON-вызовы.\n"

"ПРАВИЛО БЕЗОПАСНОСТИ: работа только внутри 'ai_workspace'.\n"

"Запрещено менять системную архитектуру, файлы сервера (main.py и т.д.).\n\n"

"ДОСТУПНЫЕ НАВЫКИ:\n"

"1. 'write_file' — {'path': 'ai_workspace/filename.py', 'content': 'код'}\n"

"2. 'read_file' — {'path': 'ai_workspace/filename.py'}\n"

"ОТВЕЧАЙ СТРОГО В ФОРМАТЕ JSON:\n"

"{\n"

' "actions": [\n'

' {"skill": "write_file", "arguments": {"path": "ai_workspace/test.py", "content": "print(\'Hello\')"}}\n'

' ],\n'

' "developer_comment": "What was done"\n'

"}"

)

},

"Explorer": {

"prompt": (

"Ты — Эксплорер, финальный агент цепочки.\n"

"Изучи отчет о выполнении навыков Кодера и подготовь финальный ответ пользователю.\n"

"Отвечай на русском языке, подтверждая статус операции."

)

}

}

async def query_llm(agent_name: str, system_prompt: str, user_content: str) -> str:

# Запрос к LLM. Спим перед вызовом, чтобы не ловить лимиты

async with api_limiter:

try:

delay = BASE_API_DELAY + random.uniform(0.5, 2.0)

logger.info(f"Пауза перед запросом к [{agent_name}]: {delay:.2f} сек...")

await asyncio.sleep(delay)

response = await client.chat.completions.create(

model=MODEL_NAME,

messages=[

{"role": "system", "content": system_prompt},

{"role": "user", "content": user_content}

],

temperature=0.15,

)

return response.choices[0].message.content.strip()

except Exception as e:

logger.error(f"Ошибка вызова OpenAI API для {agent_name}: {e}")

return ""

def fix_llm_json(raw_text: str) -> str:

# Вырезаем JSON из маркдауна, если моделька опять вернула его в ```json ... ```

text = raw_text.strip()

if "```" in text:

start_idx = text.find("{")

end_idx = text.rfind("}")

if start_idx != -1 and end_idx != -1:

return text[start_idx:end_idx + 1]

return text

async def run_sequential_pipeline(user_request: str) -> Dict[str, Any]:

logger.info(f"Запуск пайплайна под задачу: '{user_request[:50]}...'")

# 1. Планирование задач

planner_raw = await query_llm("Planner", AGENTS_CONFIG["Planner"]["prompt"], user_request)

try:

planner_data = json.loads(fix_llm_json(planner_raw))

task_for_coder = planner_data.get("task_description", user_request)

except Exception as e:

logger.warning(f"Не распарсили JSON Планёра ({e}). Юзаем сырой текст запроса.")

task_for_coder = user_request

# 2. Написание кода и выполнение функций

coder_raw = await query_llm("Coder", AGENTS_CONFIG["Coder"]["prompt"], f"ТЗ: {task_for_coder}")

skills_log = []

coder_comment = coder_raw

try:

coder_data = json.loads(fix_llm_json(coder_raw))

coder_comment = coder_data.get("developer_comment", "")

for action in coder_data.get("actions", []):

skill_name = action.get("skill")

args = action.get("arguments", {})

logger.info(f"Вызов инструмента '{skill_name}' для пути '{args.get('path')}'")

res = execute_skill(skill_name, args)

skills_log.append({

"skill": skill_name,

"target": args.get("path"),

"result": res

})

except Exception as e:

logger.warning(f"Не смогли разобрать JSON Кодера ({e}). Передаем как текст.")

# 3. Сборка финального ответа

explorer_input = (

f"1. Исходный запрос: {user_request}\n"

f"2. Комментарий Кодера: {coder_comment}\n"

f"3. Лог выполнения навыков:\n"

f"{json.dumps(skills_log, ensure_ascii=False, indent=2)}"

)

final_reply = await query_llm("Explorer", AGENTS_CONFIG["Explorer"]["prompt"], explorer_input)

return {

"status": "success",

"coder_actions_count": len(skills_log),

"skills_log": skills_log,

"reply": final_reply

}

u/app.post("/api/agent")

async def handle_agent_pipeline(request: AgentQuery):

return await run_sequential_pipeline(request.prompt)

async def run_tg_bot_worker(token: str, name: str, role: str):

bot = Bot(token=token)

dp = Dispatcher()

u/dp.message()

async def handle_message(message: types.Message):

# Реагирует только Планнер, ботов и пустые сообщения скипаем

if message.from_user.is_bot or role != "Planner" or not message.text:

return

logger.info(f"[{name}] Message: '{message.text[:50]}'")

status_msg = await message.answer("🔄 *Запрос принят.* Активирую конвейер агентов...", parse_mode="Markdown")

result = await run_sequential_pipeline(message.text)

reply = result.get("reply", "").strip()

if not reply:

reply = "⚠️ *Ошибка обработки запроса.*\n\nНе удалось получить ответ от агентов."

try:

await status_msg.delete()

except Exception:

pass

try:

await message.answer(reply, parse_mode="Markdown")

except TelegramBadRequest:

await message.answer(reply)

try:

logger.info(f"Старт polling для бота: [{name}] ({role})")

await dp.start_polling(bot)

except asyncio.CancelledError:

logger.info(f"Воркер бота [{name}] остановлен.")

except Exception as e:

logger.error(f"Ошибка пуллинга бота [{name}]: {e}")

finally:

await bot.session.close()

u/app.post("/api/start_bot")

async def start_bot_endpoint(config: BotConfig):

if config.role not in AGENTS_CONFIG:

raise HTTPException(status_code=400, detail=f"Неверная роль. Доступные: {list(AGENTS_CONFIG.keys())}")

token = config.token or TELEGRAM_TOKENS.get(config.role)

if not token or "YOUR_" in token:

raise HTTPException(status_code=400, detail="Токен пустой или дефолтный.")

if token in bot_tasks:

return {"status": "already_running", "message": f"Бот '{config.name}' уже работает."}

task = asyncio.create_task(run_tg_bot_worker(token, config.name, config.role))

bot_tasks[token] = task

task.add_done_callback(lambda t: bot_tasks.pop(token, None))

return {"status": "success", "message": f"Бот '{config.name}' успешно запущен в фоне."}

u/app.get("/")

def health_check():

return {"status": "online", "system": "Yasteria Multi-Agent Layer"}