Разработка системы обучения AI-агента на данных предыдущего сотрудника

Обучение AI-агента на данных сотрудников и корпоративной базе знаний

Корпоративный AI-агент, обученный на внутренних данных, знает контекст компании: стандарты документов, исторические решения, процессы, терминологию отрасли. В отличие от общего LLM, такой агент даёт ответы, специфичные для вашей организации, а не обобщённые рекомендации из интернета.

Подходы к обучению

RAG (Retrieval-Augmented Generation) — самый распространённый: индексируем документы в векторную БД, при запросе извлекаем релевантные фрагменты. Не требует дообучения модели.

Fine-tuning — дообучение на корпоративных диалогах и документах. Модель усваивает стиль, терминологию, типичные форматы ответов.

Гибридный подход — файнтюн на стиле + RAG на актуальных знаниях. Оптимален для production.

Сбор и подготовка корпоративных данных

from pathlib import Path
from typing import Generator
import json

class CorporateDataCollector:
    """Сбор данных из корпоративных источников"""

    async def collect_from_confluence(self, space_keys: list[str]) -> list[dict]:
        """Страницы Confluence"""
        docs = []
        for space in space_keys:
            pages = await confluence_client.get_all_pages(space)
            for page in pages:
                content = await confluence_client.get_page_content(page["id"])
                docs.append({
                    "source": "confluence",
                    "id": page["id"],
                    "title": page["title"],
                    "content": html_to_text(content),
                    "updated_at": page["version"]["when"],
                    "labels": page.get("labels", []),
                    "space": space,
                })
        return docs

    async def collect_from_email_threads(
        self,
        email_accounts: list[str],
        filter_subjects: list[str] = None,
        anonymize_pii: bool = True,
    ) -> list[dict]:
        """Email-переписка как обучающие данные для диалогов"""
        threads = []
        for account in email_accounts:
            emails = await gmail_client.get_threads(account, filter_subjects)
            for thread in emails:
                if len(thread["messages"]) >= 2:
                    # Преобразуем переписку в формат диалога
                    dialog = self.format_as_dialog(thread["messages"])
                    if anonymize_pii:
                        dialog = await self.anonymize_pii(dialog)
                    threads.append(dialog)
        return threads

    async def collect_from_tickets(
        self,
        jira_project: str,
        status: str = "Done",
        limit: int = 5000,
    ) -> list[dict]:
        """Решённые тикеты как Q&A пары"""
        tickets = await jira_client.get_issues(
            jql=f"project={jira_project} AND status={status}",
            fields=["summary", "description", "comments", "resolution"],
            limit=limit,
        )

        qa_pairs = []
        for ticket in tickets:
            if ticket.get("comments"):
                qa_pairs.append({
                    "question": f"{ticket['summary']}\n{ticket.get('description', '')[:500]}",
                    "answer": self.extract_resolution(ticket),
                    "source": "jira",
                    "ticket_id": ticket["id"],
                })

        return qa_pairs

Подготовка данных для Fine-tuning

class FinetuningDatasetBuilder:

    async def build_instruction_dataset(
        self,
        raw_docs: list[dict],
        qa_pairs: list[dict],
        target_format: str = "openai",  # "openai", "alpaca", "sharegpt"
    ) -> list[dict]:

        dataset = []

        # Из документов — генерируем Q&A через LLM
        for doc in raw_docs:
            qa_from_doc = await self.generate_qa_from_document(doc["content"])
            for qa in qa_from_doc:
                if target_format == "openai":
                    dataset.append({
                        "messages": [
                            {"role": "system", "content": "Ты — корпоративный ассистент компании. Отвечай на вопросы сотрудников."},
                            {"role": "user", "content": qa["question"]},
                            {"role": "assistant", "content": qa["answer"]},
                        ]
                    })

        # Из тикетов — готовые пары
        for qa in qa_pairs:
            if target_format == "openai":
                dataset.append({
                    "messages": [
                        {"role": "system", "content": "Ты — ассистент технической поддержки."},
                        {"role": "user", "content": qa["question"]},
                        {"role": "assistant", "content": qa["answer"]},
                    ]
                })

        # Дедупликация и фильтрация
        dataset = self.deduplicate(dataset)
        dataset = self.filter_quality(dataset, min_answer_length=50)

        return dataset

    async def generate_qa_from_document(self, document_text: str) -> list[dict]:
        """Генерирует Q&A пары из документа"""
        response = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""Создай 5-10 вопросов и ответов из следующего документа.
Вопросы должны быть такими, как их задают реальные сотрудники.
Ответы — полными и точными.

Документ:
{document_text[:3000]}

Верни JSON: [{{"question": "...", "answer": "..."}}]"""
            }],
        )
        return json.loads(response.choices[0].message.content)

    def filter_quality(self, dataset: list[dict], min_answer_length: int) -> list[dict]:
        """Фильтрует данные низкого качества"""
        filtered = []
        for item in dataset:
            messages = item.get("messages", [])
            assistant_msg = next((m for m in messages if m["role"] == "assistant"), None)
            if assistant_msg and len(assistant_msg["content"]) >= min_answer_length:
                filtered.append(item)
        return filtered

Гибридная архитектура: Fine-tune + RAG

from sentence_transformers import SentenceTransformer
from openai import OpenAI
from qdrant_client import QdrantClient

class HybridCorporateAgent:
    """Объединяет файнтюн-модель со стилем компании и RAG с актуальными знаниями"""

    def __init__(self):
        # Файнтюн-модель знает стиль и терминологию компании
        self.finetuned_client = OpenAI(base_url="http://vllm-server:8000/v1")
        self.finetuned_model = "company-assistant-ft-v2"

        # RAG для актуальных документов
        self.embed_model = SentenceTransformer("BAAI/bge-m3")
        self.vector_db = QdrantClient(host="qdrant-server")

    async def answer(self, question: str, user_context: dict = None) -> dict:
        # Шаг 1: Поиск релевантных документов
        query_embedding = self.embed_model.encode(question)
        relevant_docs = self.vector_db.search(
            collection_name="corporate_docs",
            query_vector=query_embedding,
            limit=5,
            score_threshold=0.6,
            query_filter=self.build_access_filter(user_context),  # Права доступа
        )

        # Шаг 2: Формирование контекста
        context = "\n\n".join([
            f"[{doc.payload['title']}]: {doc.payload['content']}"
            for doc in relevant_docs
        ])

        # Шаг 3: Ответ файнтюн-моделью с RAG-контекстом
        response = self.finetuned_client.chat.completions.create(
            model=self.finetuned_model,
            messages=[{
                "role": "system",
                "content": f"Ты — корпоративный ассистент. Используй документы как источник истины.\n\nДокументы:\n{context}"
            }, {
                "role": "user",
                "content": question,
            }],
            temperature=0.1,
        )

        return {
            "answer": response.choices[0].message.content,
            "sources": [{"title": d.payload["title"], "score": d.score} for d in relevant_docs],
        }

    def build_access_filter(self, user_context: dict):
        """Фильтрация по правам доступа — сотрудник видит только свои документы"""
        if not user_context:
            return None

        department = user_context.get("department", "all")
        clearance = user_context.get("clearance", "public")

        return {
            "must": [
                {"key": "access_level", "match": {"any": [clearance, "public"]}},
                {"key": "departments", "match": {"any": [department, "all"]}},
            ]
        }

Практический кейс: IT-компания, 300 сотрудников

Данные для обучения: 8 000 страниц Confluence, 12 000 решённых тикетов Jira, 5 лет email-переписки (анонимизированная).

Процесс:

1. Сбор и очистка: 3 недели (основное время — quality filtering)
2. Генерация synthetic Q&A из Confluence: 45 000 пар
3. Fine-tuning GPT-4o-mini: датасет 60 000 примеров, 3 эпохи
4. RAG-индексирование всех документов в Qdrant
5. Гибридный агент в production

Результаты:

• Точность ответов на корпоративные процессы (оценка на 500 вопросов): 91% vs 67% у базового GPT-4o
• Использование правильной корпоративной терминологии: 97% vs 43%
• Ответы на вопросы "как у нас принято делать X": только файнтюн-модель отвечает корректно
• Снижение тикетов в техподдержку: -34%

Сроки

• Сбор и очистка корпоративных данных: 2–4 недели
• Генерация synthetic Q&A: 1–2 недели
• Fine-tuning (GPU-время): 2–5 дней
• RAG-индексирование и настройка: 1–2 недели
• Тестирование и калибровка: 2 недели
• Итого: 8–13 недель