1. Qwen-Agent: архитектура и возможности

Qwen-Agent — это опенсорсный фреймворк от Alibaba Cloud для построения LLM-агентов, построенный вокруг четырёх ключевых абстракций: Agent (интеллектуальное ядро с ReAct-циклом), Tool (инструменты: от поиска до выполнения кода), LLM (единый интерфейс к разным бэкендам: DashScope, OpenAI, vLLM) и Memory (контекстное окно и долговременная память через векторные хранилища). Фреймворк поддерживает как синхронный, так и асинхронный режимы, streaming-ответы и function-calling из коробки.

Ключевое преимущество Qwen-Agent — тесная интеграция с моделями семейства Qwen (Qwen2.5, Qwen3), которые показывают state-of-the-art результаты в function-calling бенчмарках. При этом фреймворк не привязан жёстко к моделям Qwen: любой OpenAI-совместимый эндпоинт работает через универсальный класс OpenAI.

# Установка: локальный запуск с Qwen-моделью через transformers
pip install qwen-agent[gui] transformers torch accelerate

from qwen_agent.agents import Assistant
from qwen_agent.llm import get_chat_model

# Конфигурация LLM: можно переключить на vLLM, DashScope или OpenAI
llm_cfg = {
    'model': 'Qwen/Qwen2.5-7B-Instruct',
    'model_server': 'transformers',  # или 'openai', 'dashscope', 'vllm'
    'generate_cfg': {'max_tokens': 2048, 'temperature': 0.1},
}

# Создание агента с дефолтными инструментами (code_interpreter, search)
bot = Assistant(llm=llm_cfg, function_list=['code_interpreter', 'web_search'])

# Асинхронный стриминг ответа
messages = [{'role': 'user', 'content': 'Напиши Python-скрипт для парсинга курсов валют с ЦБ РФ'}]
async for response in bot.run(messages=messages):
    print(response)

2. Multi-agent оркестрация: GroupChat, Router и цепочки

Для сложных сценариев Qwen-Agent предоставляет GroupChat — паттерн, где несколько специализированных агентов общаются в общем чате, а менеджер (обычно Assistant-агент с routing-логикой) решает, кому передать слово. Это позволяет собрать конвейер: Researcher собирает информацию, Coder пишет код, Reviewer проверяет результат. Альтернативный подход — Router, который классифицирует запрос и направляет его нужному агенту без многораундового обсуждения, что снижает latency для простых запросов.

На практике production-решения часто комбинируют оба подхода: Router для быстрой диспетчеризации типовых запросов (FAQ, простые генерации) и GroupChat для комплексных задач, требующих координации нескольких агентов (исследование рынка, multi-step data pipeline).

from qwen_agent.agents import Assistant, GroupChat, Router
from qwen_agent.llm import get_chat_model

# LLM-конфигурация для всех агентов
llm_cfg = {
    'model': 'qwen-plus',
    'model_server': 'dashscope',
    'api_key': 'your-dashscope-api-key',
}

# Специализированные агенты
researcher = Assistant(
    llm=llm_cfg,
    name='Researcher',
    function_list=['web_search'],
    system_message='Ты исследователь. Ищи информацию, анализируй и докладывай.',
)

coder = Assistant(
    llm=llm_cfg,
    name='Coder',
    function_list=['code_interpreter'],
    system_message='Ты программист. Пиши и запускай Python-код для анализа данных.',
)

reviewer = Assistant(
    llm=llm_cfg,
    name='Reviewer',
    system_message='Ты ревьюер. Проверяй вывод Researcher и Coder, давай финальное резюме.',
)

# GroupChat с авто-маршрутизацией между агентами
group_chat = GroupChat(
    llm=llm_cfg,
    agents=[researcher, coder, reviewer],
    max_round=6,  # максимум 6 раундов обсуждения
)

# Альтернатива: Router для быстрой диспетчеризации
router = Router(
    llm=llm_cfg,
    agents={
        'research': researcher,
        'coding': coder,
        'review': reviewer,
    },
)

# Запуск GroupChat
response = group_chat.run(
    messages=[{'role': 'user', 'content': 'Исследуй рынок GPU-ускорителей, напиши сравнительный анализ и проверь выводы.'}]
)
for r in response:
    print(f"[{r['role']}] {r['content'][:200]}...\n---\n")

3. Кастомные инструменты: веб-скрапинг, API-вызовы и собственные Tool-классы

Стандартных инструментов редко хватает для production-задач. Qwen-Agent позволяет легко расширять функциональность через наследование от BaseTool. Достаточно определить имя, описание (его видит LLM при function-calling) и параметры в формате JSON Schema — фреймворк автоматически зарегистрирует инструмент и передаст его модели.

В production-окружении критически важно добавлять обработку ошибок, таймауты и retry-логику прямо в call()-метод инструмента. Ниже — реальный пример кастомного инструмента для работы с Jira API, который можно использовать в CI/CD-пайплайнах.

import httpx, json
from qwen_agent.tools.base import BaseTool, register_tool
from typing import Dict, Union

# Регистрируем инструмент в глобальном реестре
@register_tool('jira_tool', allow_overwrite=True)
class JiraTool(BaseTool):
    description = 'Работа с Jira: создание задач, поиск по статусу, получение деталей тикета.'
    parameters = [
        {
            'name': 'action',
            'type': 'string',
            'description': 'Действие: create_issue, search_issues, get_issue',
            'enum': ['create_issue', 'search_issues', 'get_issue'],
            'required': True,
        },
        {
            'name': 'params',
            'type': 'string',
            'description': 'JSON-строка с параметрами (project_key, summary, jql, issue_key)',
            'required': True,
        },
    ]

    def __init__(self, cfg: Dict = None):
        super().__init__(cfg)
        self.base_url = cfg.get('jira_url', 'https://your-domain.atlassian.net')
        self.email = cfg['jira_email']
        self.token = cfg['jira_api_token']
        self.client = httpx.Client(timeout=30, auth=(self.email, self.token))

    def call(self, params: Union[str, dict], **kwargs) -> str:
        # Парсим параметры (могут прийти как JSON-строка от LLM)
        if isinstance(params, str):
            params = json.loads(params)
        action = params['action']
        payload = json.loads(params['params']) if isinstance(params['params'], str) else params['params']

        if action == 'create_issue':
            resp = self.client.post(
                f'{self.base_url}/rest/api/3/issue',
                json={'fields': {'project': {'key': payload['project_key']},
                                    'summary': payload['summary'],
                                    'issuetype': {'name': 'Task'}}}
            )
        elif action == 'search_issues':
            resp = self.client.get(
                f'{self.base_url}/rest/api/3/search',
                params={'jql': payload['jql'], 'maxResults': 5}
            )
        else:
            resp = self.client.get(
                f'{self.base_url}/rest/api/3/issue/{payload['issue_key']}'
            )
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2, ensure_ascii=False)

# Использование кастомного инструмента
bot = Assistant(llm=llm_cfg, function_list=['jira_tool', 'code_interpreter'])
response = list(bot.run(messages=[{
    'role': 'user',
    'content': 'Найди все незакрытые баги в проекте DEV и создай сводный отчёт.'
}]))

4. RAG с Qwen-Agent: векторные БД и retrieval tools

Qwen-Agent интегрируется с векторными хранилищами через кастомные инструменты. Типовой RAG-пайплайн: документы → чанкинг → эмбеддинги (через dashscope.TextEmbedding или локальные модели типа bge-large) → загрузка в векторную БД (ChromaDB, Milvus, Qdrant) → retrieval tool для агента. При запросе агент вызывает retrieval tool, получает релевантные чанки и использует их как контекст для генерации ответа.

Для production важно настроить гибридный поиск (dense + sparse) и реранкинг. В примере ниже используется ChromaDB как легковесное решение для старта, но архитектура позволяет легко переключиться на Milvus для масштабирования на миллионы документов.

pip install chromadb dashscope tiktoken

import chromadb
from dashscope import TextEmbedding
from qwen_agent.tools.base import BaseTool, register_tool

@register_tool('knowledge_base')
class KnowledgeBaseTool(BaseTool):
    description = 'Поиск по внутренней базе знаний компании. Возвращает релевантные документы.'
    parameters = [{
        'name': 'query',
        'type': 'string',
        'description': 'Поисковый запрос на естественном языке',
        'required': True,
    }]

    def __init__(self, cfg: dict = None):
        super().__init__(cfg)
        self.client = chromadb.PersistentClient(path='./chroma_db')
        self.collection = self.client.get_or_create_collection('company_docs')

    def _embed(self, text: str) -> list:
        resp = TextEmbedding.call(
            model='text-embedding-v3', input=text,
            api_key=self.cfg.get('dashscope_api_key')
        )
        return resp.output['embeddings'][0]['embedding']

    def call(self, params: str, **kwargs) -> str:
        import json
        if isinstance(params, str):
            params = json.loads(params)
        query_embedding = self._embed(params['query'])
        results = self.collection.query(
            query_embeddings=[query_embedding], n_results=5,
            include=['documents', 'metadatas']
        )
        # Форматируем найденные чанки как контекст
        chunks = '\n---\n'.join(
            f"[Источник: {m.get('source', 'N/A')}]\n{d[:500]}"
            for d, m in zip(results['documents'][0], results['metadatas'][0])
        )
        return f'Найдено документов: {len(results['documents'][0])}\n\n{chunks}'

# Агент с RAG-инструментом
rag_bot = Assistant(
    llm=llm_cfg,
    function_list=['knowledge_base'],
    system_message='Отвечай строго на основе полученных документов. Если информации нет — скажи об этом.',
)

5. Docker-деплой: от Dockerfile до docker-compose с vLLM

Для production-развёртывания Qwen-Agent контейнеризация обязательна. Типовая архитектура включает три сервиса: vLLM-сервер (хостинг модели с continuous batching), Agent API (FastAPI-обёртка над Qwen-Agent с WebSocket для streaming) и Redis (кэширование, очереди, хранение сессий). Nginx выступает как reverse proxy с TLS-терминацией.

Важный нюанс: vLLM требует GPU и значительного объёма RAM. Для production используйте модели с квантованием AWQ/GPTQ (в 2–4 раза меньше VRAM при минимальной потере качества). В примере ниже vLLM запускается на порту 8000, а Agent API подключается к нему как к OpenAI-совместимому эндпоинту.

# Dockerfile для Agent API
FROM python:3.11-slim

WORKDIR /app

# Системные зависимости
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl build-essential && rm -rf /var/lib/apt/lists/*

# Python-зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Код приложения
COPY src/ ./src/

# Healthcheck
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD curl -f http://localhost:8080/health || exit 1

EXPOSE 8080
CMD ["python", "-m", "src.main"]

# docker-compose.yml — полный production-стек
version: '3.8'

services:
  vllm:
    image: vllm/vllm-openai:latest
    runtime: nvidia
    environment:
      - HF_HOME=/model-cache
    volumes:
      - ./models:/model-cache
    command: >
      --model Qwen/Qwen2.5-14B-Instruct-AWQ
      --max-model-len 8192
      --gpu-memory-utilization 0.92
      --enable-prefix-caching
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

  agent-api:
    build: .
    ports:
      - "8080:8080"
    environment:
      - QWEN_LLM_BASE_URL=http://vllm:8000/v1
      - QWEN_MODEL_NAME=Qwen/Qwen2.5-14B-Instruct-AWQ
      - REDIS_URL=redis://redis:6379/0
      - MAX_CONCURRENT_REQUESTS=16
    depends_on:
      vllm:
        condition: service_healthy
      redis:
        condition: service_started

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certs:/etc/nginx/certs:ro

volumes:
  redis_data:

6. Мониторинг и продакшен-практики: логи, retry, rate limiting, graceful shutdown

Production-эксплуатация LLM-агентов требует особого внимания к observability: каждый вызов LLM, каждый tool-вызов и каждая ошибка должны логироваться структурированно (JSON-логи — стандарт). Обязательные компоненты: retry с exponential backoff для API-вызовов (сетевые сбои, 429 от провайдера), rate limiting на уровне API Gateway (предотвращает перегрузку LLM-бэкенда), graceful shutdown (корректное завершение in-flight запросов при деплое) и circuit breaker (автоматическое отключение проблемного провайдера).

Для мониторинга рекомендуется связка Prometheus + Grafana с кастомными метриками: latency перцентили (p50/p95/p99), количество tool-вызовов на запрос, hit rate RAG, доля успешных/проваленных LLM-вызовов. Алерты — на рост p99 latency выше порога и на падение success rate ниже 95%.

# src/main.py — FastAPI + мониторинг + graceful shutdown
import asyncio, time, signal, logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from prometheus_fastapi_instrumentator import Instrumentator
from tenacity import retry, stop_after_attempt, wait_exponential

# Структурированное логирование
logging.basicConfig(
    level=logging.INFO,
    format='{"ts":"%(asctime)s","lvl":"%(levelname)s","msg":"%(message)s"},"ctx":%(otel_context)s}',
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger('qwen-agent-api')

# Rate limiting: 30 запросов в минуту с IP
limiter = Limiter(key_func=get_remote_address, default_limits=["30/minute"])

# Graceful shutdown: отслеживаем активные запросы
active_requests = 0
shutdown_event = asyncio.Event()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Запуск: инициализация агента
    logger.info('Initializing Qwen-Agent pipeline...')
    app.state.agent = Assistant(llm=llm_cfg, function_list=['code_interpreter'])
    logger.info('Agent ready')
    yield
    # Shutdown: ждём завершения активных запросов (макс 30 сек)
    global shutdown_event
    shutdown_event.set()
    logger.info(f'Shutting down, {active_requests} active requests...')
    for _ in range(30):
        if active_requests == 0:
            break
        await asyncio.sleep(1)
    logger.info('Shutdown complete')

app = FastAPI(lifespan=lifespan, title='Qwen Agent API')
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

# Prometheus-метрики: latency, requests, errors
Instrumentator().instrument(app).expose(app, endpoint='/metrics')

@app.post('/chat')
@limiter.limit("30/minute")
async def chat(request: Request):
    global active_requests
    if shutdown_event.is_set():
        from fastapi.responses import JSONResponse
        return JSONResponse({'error': 'Server is shutting down'}, status_code=503)

    active_requests += 1
    t0 = time.monotonic()
    try:
        body = await request.json()
        logger.info(f'Processing chat request: {body.get('messages')[-1]['content'][:100]}...')
        # Retry-логика для LLM-вызовов
        @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
        async def _run_agent(msgs):
            result = []
            async for r in request.app.state.agent.run(messages=msgs):
                result.append(r)
            return result
        response = await _run_agent(body['messages'])
        elapsed = time.monotonic() - t0
        logger.info(f'Request completed in {elapsed:.2f}s')
        return {'response': response, 'elapsed_s': round(elapsed, 2)}
    except Exception as e:
        logger.error(f'Request failed: {e}')
        from fastapi.responses import JSONResponse
        return JSONResponse({'error': str(e)}, status_code=500)
    finally:
        active_requests -= 1

@app.get('/health')
async def health():
    return {'status': 'ok', 'active_requests': active_requests}

7. Сравнение с AutoGen, CrewAI и LangGraph

Выбор фреймворка для агентов — стратегическое решение. Ниже — честное сравнение по ключевым критериям, основанное на реальном опыте внедрения каждого инструмента в production-среду.

Вердикт: Qwen-Agent — оптимальный выбор, если вы работаете с моделями Qwen, разворачиваетесь в Alibaba Cloud или вам критично качество function-calling. Для западного рынка с OpenAI-моделями LangGraph даёт больше гибкости. CrewAI хорош для быстрого прототипирования бизнес-процессов с role-playing агентами. AutoGen остаётся стандартом в исследовательской среде, но требует больше ручной настройки для production.

📋 Production-чеклист для Qwen-Agent