Создание AI-приложений стало доступнее чем когда-либо благодаря API языковых моделей и богатой экосистеме инструментов. От простого чат-бота до сложной RAG системы – понимание правильного подхода и архитектурных паттернов поможет быстро перейти от идеи к работающему продукту1.

Определение типа AI-приложения

Первый шаг – понять какой тип приложения вы создаете. От этого зависит архитектура и выбор технологий.

Тип приложения Описание Примеры Сложность
Simple chat bot Прямое взаимодействие с LLM Поддержка клиентов, FAQ бот ⭐☆☆☆☆
RAG система LLM + база знаний Документация ассистент, поиск по контенту ⭐⭐⭐☆☆
AI агенты Автономные действия, function calling Автоматизация задач, workflow ⭐⭐⭐⭐☆
Мультимодальные приложения Текст + изображения/аудио/видео Анализ документов, генерация контента ⭐⭐⭐⭐☆
Fine-tuned модели Кастомизированная модель для домена Специализированные отраслевые решения ⭐⭐⭐⭐⭐

Стек технологий для AI-приложений

Backend

Python: Стандарт для AI разработки

  • FastAPI: Современный, быстрый фреймворк для API
  • LangChain: Фреймворк для работы с LLM (chains, agents)
  • LlamaIndex: Специализирован на RAG и работе с данными

TypeScript/Node.js: Альтернатива для JS разработчиков

  • LangChain.js: JS версия LangChain
  • Vercel AI SDK: Удобный для Next.js проектов

Frontend

  • React/Next.js: Наиболее популярный выбор
  • Streamlit: Быстрые прототипы Python-приложений
  • Gradio: Простые ML интерфейсы

Векторные базы данных

  • Pinecone: Managed solution, просто для старта
  • Chroma: Embedded, идеально для разработки
  • Qdrant: High-performance, open-source
  • Weaviate: Для сложных сценариев поиска

Создание простого чат-бота: пошаговое руководство

Шаг 1: Настройка окружения

# Создаем виртуальное окружение
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# Устанавливаем зависимости
pip install fastapi uvicorn openai python-dotenv

Шаг 2: Backend с FastAPI

# main.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
app = FastAPI()

# CORS для фронтенда
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class ChatRequest(BaseModel):
    message: str
    conversation_history: list = []

@app.post("/chat")
async def chat(request: ChatRequest):
    try:
        # Строим messages из истории + новое сообщение
        messages = [
            {"role": "system", "content": "Ты полезный ассистент."}
        ]
        messages.extend(request.conversation_history)
        messages.append({"role": "user", "content": request.message})
        
        # Запрос к OpenAI
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0.7,
            max_tokens=500
        )
        
        assistant_message = response.choices[0].message.content
        
        return {
            "response": assistant_message,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Шаг 3: Простой Frontend





    AI Chat Bot
    


    

Построение RAG системы

RAG (Retrieval-Augmented Generation) – один из наиболее практичных паттернов для AI приложений2.

Архитектура RAG

  1. Индексация: Документы → Chunking → Embeddings → Vector DB
  2. Retrieval: Query → Embedding → Similarity Search → Relevant Docs
  3. Generation: Query + Retrieved Docs → LLM → Answer

Реализация с LangChain

# rag_system.py
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI

# 1. Загрузка документов
loader = DirectoryLoader('./documents', glob="**/*.txt")
documents = loader.load()

# 2. Chunking
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)

# 3. Создание эмбеддингов и векторной БД
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. Создание retrieval chain
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 5. Использование
query = "Как настроить API?"
result = qa_chain.run(query)
print(result)

Продвинутые паттерны

AI Agents с function calling

Agents могут принимать решения и вызывать функции для выполнения действий:

from openai import OpenAI
import json

client = OpenAI()

# Определяем доступные функции
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Получить погоду для города",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "Название города"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Поиск в интернете",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Поисковый запрос"}
                },
                "required": ["query"]
            }
        }
    }
]

def get_weather(city):
    # Реальный API call к погодному сервису
    return f"В городе {city} сейчас 15°C, облачно"

def search_web(query):
    # Реальный поиск или API call
    return f"Результаты поиска для: {query}"

# Agent loop
messages = [
    {"role": "user", "content": "Какая погода в Москве?"}
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

# Если модель решила вызвать функцию
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    # Вызываем соответствующую функцию
    if function_name == "get_weather":
        result = get_weather(**arguments)
    elif function_name == "search_web":
        result = search_web(**arguments)
    
    # Отправляем результат обратно модели
    messages.append(response.choices[0].message)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result
    })
    
    # Финальный ответ с использованием результата функции
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    
    print(final_response.choices[0].message.content)

Streaming responses

Для лучшего UX реализуйте потоковую передачу ответов:

# FastAPI endpoint со streaming
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        stream = client.chat.completions.create(
            model="gpt-4o",
            messages=request.messages,
            stream=True
        )
        
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}\n\n"
        
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

Оптимизация производительности

Кэширование

from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def cached_embedding(text):
    # Кэшируем эмбеддинги для экономии
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# Кэширование целых ответов
import redis
r = redis.Redis()

def get_cached_response(prompt):
    cache_key = hashlib.md5(prompt.encode()).hexdigest()
    cached = r.get(cache_key)
    
    if cached:
        return cached.decode()
    
    # Если нет в кэше - делаем запрос
    response = client.chat.completions.create(...)
    result = response.choices[0].message.content
    
    # Сохраняем в кэш на 1 час
    r.setex(cache_key, 3600, result)
    
    return result

Batch processing

Обрабатывайте множество запросов параллельно:

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def process_multiple_queries(queries):
    tasks = [
        async_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}]
        )
        for query in queries
    ]
    
    responses = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in responses]

# Использование
queries = ["Вопрос 1", "Вопрос 2", "Вопрос 3"]
results = asyncio.run(process_multiple_queries(queries))

Тестирование AI приложений

Unit тесты с мокированием

# test_chatbot.py
import pytest
from unittest.mock import Mock, patch

@patch('openai.OpenAI')
def test_chat_endpoint(mock_openai):
    # Настраиваем mock
    mock_response = Mock()
    mock_response.choices = [
        Mock(message=Mock(content="Test response"))
    ]
    mock_response.usage = Mock(
        prompt_tokens=10,
        completion_tokens=20,
        total_tokens=30
    )
    
    mock_openai.return_value.chat.completions.create.return_value = mock_response
    
    # Тестируем функцию
    from main import chat
    result = chat(ChatRequest(message="Test"))
    
    assert result["response"] == "Test response"
    assert result["usage"]["total_tokens"] == 30

Evaluation метрики

Для RAG систем важно измерять качество:

  • Retrieval metrics: Precision, Recall, MRR
  • Generation metrics: BLEU, ROUGE для сравнения с эталоном
  • End-to-end: Human evaluation, LLM-as-judge

Deployment

Опции для деплоя

Платформа Преимущества Лучше для
Vercel Простота, бесплатный tier, отлично для Next.js Фронтенд + serverless functions
Railway Легкий деплой, поддержка БД Полноценные backend приложения
AWS / GCP / Azure Полный контроль, масштабируемость Production enterprise приложения
Docker + любой хостинг Портируемость, изоляция Кастомные требования

Docker setup

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Распространенные ошибки и как их избежать

Ошибка 1: Игнорирование управления промптами

Проблема: Промпты захардкожены в коде

Решение: Используйте prompt templates и храните их отдельно:

# prompts.yaml
system_prompts:
  customer_support: |
    Ты вежливый и профессиональный агент поддержки.
    Отвечай кратко и по делу.
    Если не знаешь ответ - честно признайся.
  
  code_helper: |
    Ты опытный программист.
    Предоставляй код с комментариями.
    Объясняй сложные концепции простыми словами.

Ошибка 2: Отсутствие мониторинга

Решение: Логируйте все запросы, латенси, costs:

import logging
import time

def log_request(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        duration = time.time() - start
        
        logging.info({
            "function": func.__name__,
            "duration": duration,
            "tokens": result.get("usage", {}).get("total_tokens"),
            "cost": calculate_cost(result)
        })
        
        return result
    return wrapper

Ошибка 3: Недооценка costs

Решение: Внедрите бюджетные лимиты с самого начала:

daily_budget = 100  # $100 в день
current_spend = 0

def check_budget():
    if current_spend >= daily_budget:
        raise Exception("Daily budget exceeded")

@app.post("/chat")
async def chat(request):
    check_budget()
    # ... остальной код
    current_spend += cost

Ключевые выводы

  • Начните с простого чат-бота, затем постепенно добавляйте complexity
  • RAG системы - практичный паттерн для добавления знаний к LLM
  • Используйте established фреймворки (LangChain, LlamaIndex) для ускорения разработки
  • Кэширование, batch processing и streaming критичны для production
  • Мониторинг costs и performance с первого дня экономит деньги и проблемы

Примечания и источники

  1. Liu, J., et al. (2023). LLM-Based Applications: Survey and Insights. arXiv.
  2. Lewis, P., et al. (2020). Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. NeurIPS.