AI-система управления знаниями

AI-система управления знаниями

Управление знаниями разваливается по одной причине: их накопление требует усилий, которых у людей нет. Написать статью после решения проблемы — это дополнительные 30–60 минут поверх основной работы. В результате знания остаются в головах, в Slack-переписках, в личных заметках — и уходят вместе с сотрудниками.

AI-система управления знаниями переворачивает процесс: вместо того чтобы просить людей документировать, система сама извлекает знания из существующих потоков — переписок, тикетов, звонков, commit messages — и структурирует их без дополнительных усилий команды.

Архитектура: автоматическое извлечение знаний

from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Qdrant
from sentence_transformers import SentenceTransformer
from datetime import datetime
import json

class KnowledgeExtractionPipeline:
    """Извлекает знания из неструктурированных источников"""

    EXTRACTION_PROMPT = """Проанализируй текст и извлеки структурированное знание.

Текст (источник: {source}):
{text}

Определи:
1. Тип знания: решение_проблемы | best_practice | процесс | определение | кейс
2. Заголовок (до 10 слов)
3. Суть знания (2–4 предложения, только факты)
4. Условия применимости (когда это знание актуально)
5. Связанные темы/теги
6. Уверенность в качестве (0–1): насколько текст содержит реальное знание

Верни JSON. Если знания нет (small talk, статус-апдейт) — верни null."""

    def __init__(self, llm: ChatOpenAI, vector_store: Qdrant):
        self.llm = llm
        self.vector_store = vector_store
        self.embedder = SentenceTransformer("intfloat/multilingual-e5-large")

    async def process_slack_thread(self, thread: dict) -> list[dict]:
        """Извлекает знания из Slack-треда"""
        thread_text = "\n".join([
            f"{msg['user']}: {msg['text']}"
            for msg in thread["messages"]
        ])

        result = await self.llm.ainvoke(
            self.EXTRACTION_PROMPT.format(
                source=f"Slack #{thread['channel']}",
                text=thread_text[:3000]
            )
        )

        try:
            knowledge = json.loads(result.content)
            if knowledge and knowledge.get("confidence", 0) >= 0.7:
                return [self._store_knowledge(knowledge, thread)]
        except Exception:
            pass
        return []

    async def process_jira_ticket(self, ticket: dict) -> list[dict]:
        """Извлекает знание из решённого тикета"""
        if ticket["status"] != "Done":
            return []

        text = f"""Проблема: {ticket['title']}
Описание: {ticket.get('description', '')}
Комментарии: {' '.join([c['body'] for c in ticket.get('comments', [])])}
Решение: {ticket.get('resolution', '')}"""

        return await self._extract_and_store(text, f"Jira {ticket['key']}")

Граф знаний: связи между концепциями

Разрозненные статьи — слабая база знаний. Граф знаний связывает концепции и позволяет отвечать на вопросы типа «что ещё связано с этой проблемой?»

import networkx as nx
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class KnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.node_embeddings = {}

    def add_knowledge_node(self, knowledge_id: str, knowledge: dict, embedding: np.ndarray):
        self.graph.add_node(knowledge_id, **knowledge)
        self.node_embeddings[knowledge_id] = embedding

        # Автоматически строим связи с семантически близкими узлами
        self._auto_link(knowledge_id, embedding, threshold=0.75)

    def _auto_link(self, new_id: str, new_emb: np.ndarray, threshold: float):
        if len(self.node_embeddings) < 2:
            return

        existing_ids = [k for k in self.node_embeddings if k != new_id]
        existing_embs = np.array([self.node_embeddings[k] for k in existing_ids])

        similarities = cosine_similarity([new_emb], existing_embs)[0]

        for node_id, sim in zip(existing_ids, similarities):
            if sim >= threshold:
                self.graph.add_edge(new_id, node_id, weight=float(sim), type="related")

    def get_related(self, knowledge_id: str, depth: int = 2) -> list[str]:
        """Возвращает связанные узлы до указанной глубины"""
        if knowledge_id not in self.graph:
            return []
        return list(nx.ego_graph(self.graph, knowledge_id, radius=depth).nodes)

Автоматическое устаревание знаний

Знания устаревают. Статья о настройке VPN на старой версии ПО — хуже, чем её отсутствие: она вводит в заблуждение.

class KnowledgeFreshnessChecker:
    STALENESS_CHECK_PROMPT = """Оцени актуальность следующей статьи.

Статья (создана: {created_date}):
{content}

Последние связанные изменения в репозитории:
{recent_commits}

Определи:
1. Статус: актуально | устарело | требует_проверки
2. Причина (если устарело/требует проверки)
3. Рекомендуемое действие

Верни JSON."""

    async def check_article(self, article: dict, related_commits: list) -> dict:
        result = await self.llm.ainvoke(
            self.STALENESS_CHECK_PROMPT.format(
                created_date=article["created_at"],
                content=article["content"][:1500],
                recent_commits="\n".join([
                    f"- {c['date']}: {c['message']}"
                    for c in related_commits[:10]
                ])
            )
        )
        return json.loads(result.content)

Кейс: разработческая компания, 80 инженеров. До внедрения: 340 статей в Confluence, 60% не обновлялись более года, команда не доверяла документации. После 6 месяцев работы AI-системы: извлечено 1200+ единиц знаний из Slack-тредов и Jira-тикетов, 89 статей помечены как устаревшие и отправлены на ревью владельцам. Индекс доверия к документации (опрос команды): 2.1/5 → 3.9/5.

Интеграции

Сроки: базовый pipeline с Slack + Jira: 4–6 недель; граф знаний и проверка актуальности: 8–12 недель.