Разработка AI-ассистента для корпоративной базы знаний

Разработка AI-ассистента для корпоративной базы знаний

Корпоративная база знаний — Confluence, Notion, SharePoint, внутренние вики — содержит огромное количество информации, которую сотрудники не могут найти или не знают о существовании. AI-ассистент делает эти знания доступными через диалог: сотрудник задаёт вопрос на естественном языке и получает ответ с указанием источников.

Архитектура RAG-ассистента

from anthropic import Anthropic
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path
import json

client = Anthropic()
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

class CorporateKBAssistant:

    def __init__(self, db_path: str = "./kb_index"):
        self.vectorstore = Chroma(
            collection_name="corporate_kb",
            embedding_function=embeddings,
            persist_directory=db_path,
        )
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
        )
        self.conversation_history: dict[str, list] = {}

    def index_confluence_page(self, page: dict):
        """Индексирует страницу Confluence"""
        chunks = self.splitter.split_text(page["body"])

        self.vectorstore.add_texts(
            texts=chunks,
            metadatas=[{
                "title": page["title"],
                "space": page["space"],
                "url": page["url"],
                "updated": page["updated"],
                "author": page["author"],
            } for _ in chunks]
        )

    def search(self, query: str, k: int = 5) -> list[dict]:
        """Семантический поиск по базе знаний"""
        results = self.vectorstore.similarity_search_with_score(query, k=k)
        return [{
            "content": doc.page_content,
            "metadata": doc.metadata,
            "score": score,
        } for doc, score in results]

    def answer(self, question: str, user_id: str) -> dict:
        """Отвечает на вопрос с указанием источников"""
        # Поиск релевантных материалов
        search_results = self.search(question)

        if not search_results:
            return {"answer": "По вашему запросу ничего не найдено в базе знаний.", "sources": []}

        # Формируем контекст
        context = "\n\n".join([
            f"[{r['metadata']['title']}]:\n{r['content']}"
            for r in search_results[:4]
        ])

        # История диалога
        history = self.conversation_history.get(user_id, [])

        messages = history + [{
            "role": "user",
            "content": f"""Вопрос сотрудника: {question}

Релевантные материалы из базы знаний:
{context}

Ответь на вопрос ТОЛЬКО на основе предоставленных материалов.
Если информации недостаточно — скажи об этом прямо."""
        }]

        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            system="""Ты — корпоративный ассистент по базе знаний компании.
Отвечай точно и по существу. Всегда указывай источники.
Не придумывай информацию, которой нет в предоставленных материалах.""",
            messages=messages,
        )

        answer_text = response.content[0].text

        # Обновляем историю
        history.append({"role": "user", "content": question})
        history.append({"role": "assistant", "content": answer_text})
        self.conversation_history[user_id] = history[-10:]

        return {
            "answer": answer_text,
            "sources": [
                {
                    "title": r["metadata"]["title"],
                    "url": r["metadata"].get("url", ""),
                    "relevance": round(1 - r["score"], 2),
                }
                for r in search_results[:3]
            ],
        }

Синхронизация с Confluence

import requests
from base64 import b64encode

class ConfluenceSyncer:

    def __init__(self, base_url: str, username: str, api_token: str):
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Basic {b64encode(f'{username}:{api_token}'.encode()).decode()}",
            "Content-Type": "application/json",
        }

    def get_all_pages(self, space_key: str) -> list[dict]:
        """Получает все страницы из пространства Confluence"""
        pages = []
        start = 0

        while True:
            response = requests.get(
                f"{self.base_url}/rest/api/content",
                headers=self.headers,
                params={
                    "spaceKey": space_key,
                    "type": "page",
                    "expand": "body.storage,metadata.labels,history.lastUpdated",
                    "start": start,
                    "limit": 50,
                }
            )
            data = response.json()
            results = data.get("results", [])
            pages.extend(results)

            if len(results) < 50:
                break
            start += 50

        return [{
            "title": page["title"],
            "body": self._html_to_text(page["body"]["storage"]["value"]),
            "space": space_key,
            "url": f"{self.base_url}/wiki/spaces/{space_key}/pages/{page['id']}",
            "updated": page["history"]["lastUpdated"]["when"],
            "author": page["history"]["lastUpdated"]["by"]["displayName"],
        } for page in pages]

    def _html_to_text(self, html: str) -> str:
        """Извлекает чистый текст из HTML Confluence"""
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        return soup.get_text(separator="\n", strip=True)

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

Проблема: 45 минут в день на поиск нужной информации (опрос сотрудников). Confluence с 3000+ страниц, большинство не знали где искать.

Внедрение:

• Индексирование 3200 страниц Confluence за 4 часа
• Telegram-бот + Slack-бот для доступа
• Ежедневная синхронизация новых страниц

Метрики:

• Время поиска информации: 45 мин/день → 8 мин/день
• Обращения в Slack "кто знает где это?": -71%
• Точность ответов (оценка пользователей): 4.3/5.0
• 9% вопросов бот не смог ответить и направил к конкретным коллегам

Сроки

• Базовый RAG + Confluence синхронизация: 1 неделя
• Telegram/Slack интеграция: 3–5 дней
• Автосинхронизация + инкрементное индексирование: 1 неделя
• Мультисource (Notion + Jira + Google Drive): +1 неделя каждый