פיתוח סוכן AI המאומן על נתוני עובד שעוזב

עובד מפתח עוזב — המומחיות עוזבת איתו

מאגר הידע נשאר, אבל מציאת תשובות היא בעיה. מודל שפה גדול (LLM) רגיל לא מכיר את הטרמינולוגיה והתהליכים שלכם. סוכן AI שאומן על הנתונים של אותו עובד יכול לענות כמו שהיה עונה. אנחנו מפתחים סוכנים כאלה במפתח מלא: מאיסוף נתונים ועד הפקה.

למה מודל שפה גדול רגיל לא מספיק

מודל כללי נותן תשובות כלליות מהאינטרנט. הוא לא יודע שבחברה שלכם "אישור חוזה" עובר דרך שלוש רמות. הוא לא יודע שב-Confluence יש תבנית דוח וב-Jira יש היסטוריה של משימות דומות. דיוק בתהליכים ארגוניים ללא כוונון עדין הוא כ-67%. סוכן שאומן על הנתונים שלכם מעלה אותו ל-91%.

אילו בעיות אנחנו פותרים

• אובדן מומחיות כשעובד עוזב. הידע הייחודי שלהם על תהליכים, החלטות ואנשי קשר נשאר בראשם. סוכן AI לוכד אותו במודל ובאינדקס RAG.
• נתונים לא מובנים. 80% מהידע הארגוני נמצא ב-Confluence, Jira ומיילים. אנחנו אוספים, מנקים ומבצעים אינדוקס למסד נתונים וקטורי (Qdrant, pgvector).
• אחזור תשובות איטי. במקום 15 דקות של חיפוש ב-Confluence — שאילתה אחת לסוכן. הפחתת פניות לתמיכה ב-34%.

באיזו גישת אימון לסוכן AI לבחור?

הגישה ההיברידית היא היחידה שמבטיחה גם דיוק בסגנון וגם רעננות. אנחנו משתמשים בה בכל פרויקטי ההפקה. למידע נוסף על RAG ראו ויקיפדיה.

איך אנחנו אוספים ומכינים נתונים

מקורות נתונים: Confluence (דפים), Jira (טיקטים שנפתרו), התכתבות דוא"ל (אנונימית), קבצים ארגוניים (PDF, DOCX).

תהליך ההכנה:

1. איסוף נתונים באמצעות API (Confluence REST, Jira API, IMAP למיילים).
2. ניקוי: html-to-text, הסרת כפילויות, סינון איכות (הסרת תשובות קצרות מ-50 טוקנים).
3. יצירת שאלות ותשובות סינתטיות ממסמכים — אנו משתמשים ב-GPT-4o-mini ליצירת עד 10 זוגות לכל מסמך.
4. המרת פורמט: פורמט הודעות OpenAI (system/user/assistant).

דוגמת קוד לאיסוף נתונים:

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

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

איך הארכיטקטורה ההיברידית עובדת: כוונון עדין + RAG

המודל המכוונן מכיר את הסגנון והטרמינולוגיה של החברה. 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 עובדים

לקוח — חברת IT עם 300 עובדים — היה לו מפתח בכיר שעזב וניהל תהליכים מרכזיים. במשך 5 שנים הוא צבר 8,000 דפי Confluence והשתתף בפתרון 12,000 טיקטים ב-Jira. אספנו וניקינו נתונים ב-3 שבועות, יצרנו 45,000 זוגות שאלה-תשובה סינתטיים, כיווננו עדין את GPT-4o-mini על 60,000 דוגמאות (3 אפוקות), וטענו את כל המסמכים ל-Qdrant. תוצאות: דיוק בתהליכים ארגוניים 91% לעומת 67% עבור GPT-4o בסיסי, טרמינולוגיה נכונה 97% לעומת 43%, הפחתת פניות לתמיכה ב-34%.

מה כלול בעבודה

כמה זמן אורך הפיתוח?

בין 8 ל-13 שבועות תלוי בנפח ואיכות הנתונים. רוב הזמן מושקע באיסוף וניקוי — אי אפשר לזרז זאת בלי לפגוע באיכות. העלות נקבעת באופן פרטני לפי מספר המקורות והדיוק הנדרש.

טעויות אופייניות באימון סוכן AI

• התעלמות מהרשאות גישה. ללא סינון, עובד עלול לראות מסמכים ממחלקה אחרת. הארכיטקטורה שלנו כוללת 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 שבודק מחלקה ורמת סיווג.
• שימוש רק ב-RAG. ללא כוונון עדין, המודל לא לומד את הסגנון הארגוני — תשובות נשמעות כמו מהאינטרנט, לא כמו מעמית.
• סינון איכות חלש. אם תשובות גרועות (קצרות מ-50 טוקנים או ללא עובדות) נכנסות למערך הנתונים, האיכות יורדת. אנחנו מסננים לפי אורך מינימלי ודמיון סמנטי.
• ללא בדיקות על שאלות אמיתיות. בדיקות סינתטיות לא חושפות פערים. אנחנו בודקים על 500+ שאלות ממשתמשים עתידיים.

למה להזמין פיתוח מאיתנו?

5 שנות ניסיון ב-NLP ו-MLOps, 30+ סוכני AI פרוסים. אנחנו מבטיחים שהסוכן יענה בדיוק של לפחות 85% על התהליכים המוגדרים. אנחנו משתמשים רק בכלים מוכחים: Hugging Face, Qdrant, vLLM, ONNX Runtime. לאחר המסירה, אנו מספקים תמיכה ואימון מחדש ככל שמצטברים נתונים חדשים.

קבלו ייעוץ — ננתח את הנתונים שלכם ונציע את הגישה האופטימלית. פיתוח במפתח מלא מאפס או אינטגרציה לתשתית קיימת. הזמינו פיתוח סוכן AI — שמרו על המומחיות של הצוות שלכם.