Навчання 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-переписки (анонімізованої).
Процес:
- Збір та очистка: 3 тижні (основний час — quality filtering)
- Генерація synthetic Q&A з Confluence: 45 000 пар
- Fine-tuning GPT-4o-mini: датасет 60 000 прикладів, 3 епохи
- RAG-індексування всіх документів у Qdrant
- Гібридний агент у production
Результати:
- Точність відповідей на корпоративні процеси (оцінка на 500 питаннях): 91% vs 67% у базового GPT-4o
- Використання правильної корпоративної термінології: 97% vs 43%
- Відповіді на питання "як ми це робимо": тільки файнтюн-модель відповідає коректно
- Зменшення тикетів у техпідтримці: -34%
Терміни
- Збір та очистка корпоративних даних: 2–4 тижні
- Генерація synthetic Q&A: 1–2 тижні
- Fine-tuning (GPU-час): 2–5 днів
- RAG-індексування та налаштування: 1–2 тижні
- Тестування та калібрування: 2 тижні
- Всього: 8–13 тижнів