← Назад к курсу
Подробное учебное пособие по Model Context Protocol (MCP)
1. Теоретическая часть
1.1 Что такое Model Context Protocol (MCP)?
Model Context Protocol — это подход к управлению контекстом при работе с большими языковыми моделями (LLM). В отличие от обычной передачи контекста, MCP фокусируется на:
- Структурированном представлении контекста
- Динамическом управлении релевантностью информации
- Оптимизации использования ограниченного окна контекста моделей
1.2 Проблемы, которые решает MCP
- Ограничение длины контекста (токен-лимит)
- Потеря релевантности в длинных диалогах
- Проблема "забывания" важной информации
- Низкая эффективность использования контекста
1.3 Ключевые концепции MCP
┌─────────────────────────────────────────────────────┐ │ Model Context Protocol │ ├─────────────────────────────────────────────────────┤ │ • Контекстуальные чанки (Chunks) │ │ • Семантическое индексирование │ │ • Динамическое извлечение │ │ • Приоритезация контекста │ │ • Графы зависимостей │ └─────────────────────────────────────────────────────┘
2. Практическая реализация на Python
2.1 Базовый пример MCP
import json
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
import hashlib
@dataclass
class ContextChunk:
"""Базовый класс для чанка контекста"""
content: str
metadata: Dict[str, Any]
timestamp: datetime
importance: float # 0.0 - 1.0
chunk_id: str
def __post_init__(self):
if not self.chunk_id:
self.chunk_id = hashlib.md5(
f"{self.content}{self.timestamp}".encode()
).hexdigest()[:8]
class ModelContextProtocol:
"""Базовая реализация MCP"""
def __init__(self, max_chunks: int = 10, max_tokens: int = 4000):
self.max_chunks = max_chunks
self.max_tokens = max_tokens
self.context_chunks: List[ContextChunk] = []
self.conversation_history: List[Dict] = []
def add_context(self,
content: str,
metadata: Dict = None,
importance: float = 0.5) -> str:
"""Добавление нового контекста"""
if metadata is None:
metadata = {}
chunk = ContextChunk(
content=content,
metadata=metadata,
timestamp=datetime.now(),
importance=importance,
chunk_id=""
)
# Добавляем метаданные по умолчанию
metadata.update({
'tokens': self._estimate_tokens(content),
'source': 'user_input',
'version': 1.0
})
self.context_chunks.append(chunk)
# Управляем размером контекста
self._manage_context_size()
return chunk.chunk_id
def _manage_context_size(self):
"""Управление размером контекста"""
# Сортировка по важности и времени
self.context_chunks.sort(
key=lambda x: (x.importance, x.timestamp),
reverse=True
)
# Ограничение количества чанков
if len(self.context_chunks) > self.max_chunks:
self.context_chunks = self.context_chunks[:self.max_chunks]
# Проверка лимита токенов
total_tokens = sum(
chunk.metadata.get('tokens', 0)
for chunk in self.context_chunks
)
while total_tokens > self.max_tokens and self.context_chunks:
removed = self.context_chunks.pop()
total_tokens -= removed.metadata.get('tokens', 0)
def _estimate_tokens(self, text: str) -> int:
"""Оценка количества токенов (упрощенная)"""
# Примерная оценка: 1 токен ≈ 4 символа для английского
# Для русского можно использовать 1 токен ≈ 2 символа
return len(text) // 4
def get_relevant_context(self,
query: str = None,
top_k: int = 5) -> str:
"""Получение релевантного контекста"""
if not query or not self.context_chunks:
# Возвращаем все чанки, отсортированные по важности
sorted_chunks = sorted(
self.context_chunks,
key=lambda x: x.importance,
reverse=True
)[:top_k]
return "\n\n".join([chunk.content for chunk in sorted_chunks])
# Простой механизм релевантности
relevant_chunks = []
query_words = set(query.lower().split())
for chunk in self.context_chunks:
chunk_words = set(chunk.content.lower().split())
common_words = query_words.intersection(chunk_words)
if common_words:
relevance = len(common_words) / len(query_words)
weighted_score = relevance * chunk.importance
relevant_chunks.append((weighted_score, chunk))
# Сортировка по релевантности
relevant_chunks.sort(key=lambda x: x[0], reverse=True)
return "\n\n".join(
[chunk.content for _, chunk in relevant_chunks[:top_k]]
)
def update_importance(self, chunk_id: str, importance: float):
"""Обновление важности чанка"""
for chunk in self.context_chunks:
if chunk.chunk_id == chunk_id:
chunk.importance = max(0.0, min(1.0, importance))
break
def save_context(self, filename: str):
"""Сохранение контекста в файл"""
data = {
'chunks': [
{
'content': chunk.content,
'metadata': chunk.metadata,
'timestamp': chunk.timestamp.isoformat(),
'importance': chunk.importance,
'chunk_id': chunk.chunk_id
}
for chunk in self.context_chunks
]
}
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def load_context(self, filename: str):
"""Загрузка контекста из файла"""
with open(filename, 'r', encoding='utf-8') as f:
data = json.load(f)
self.context_chunks = []
for chunk_data in data['chunks']:
chunk = ContextChunk(
content=chunk_data['content'],
metadata=chunk_data['metadata'],
timestamp=datetime.fromisoformat(chunk_data['timestamp']),
importance=chunk_data['importance'],
chunk_id=chunk_data['chunk_id']
)
self.context_chunks.append(chunk)
# Пример использования
def basic_mcp_example():
"""Базовый пример использования MCP"""
mcp = ModelContextProtocol(max_chunks=5, max_tokens=2000)
# Добавляем контекст с разной важностью
mcp.add_context(
"Пользователь предпочитает получать ответы на русском языке.",
importance=0.9
)
mcp.add_context(
"Текущая тема обсуждения: искусственный интеллект и машинное обучение.",
importance=0.8
)
mcp.add_context(
"Пользователь интересуется нейронными сетями и глубоким обучением.",
importance=0.7
)
# Получаем релевантный контекст
query = "Расскажи о машинном обучении"
context = mcp.get_relevant_context(query=query, top_k=3)
print("Релевантный контекст:")
print(context)
print("\n" + "="*50 + "\n")
# Обновляем важность
print("Обновление важности чанков...")
# Здесь нужно знать ID чанка, в реальном сценарии
# можно добавить метод поиска по содержанию
# Сохраняем контекст
mcp.save_context("context_backup.json")
return mcp
2.2 Продвинутая реализация с семантическим поиском
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import faiss
import pickle
class AdvancedMCP(ModelContextProtocol):
"""Продвинутая реализация MCP с векторными embedding"""
def __init__(self,
max_chunks: int = 20,
max_tokens: int = 8000,
model_name: str = 'sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2'):
super().__init__(max_chunks, max_tokens)
self.embedding_model = SentenceTransformer(model_name)
self.embeddings = []
self.index = None
self._build_index()
def _build_index(self):
"""Построение FAISS индекса для быстрого поиска"""
dimension = self.embedding_model.get_sentence_embedding_dimension()
self.index = faiss.IndexFlatL2(dimension)
def add_context(self,
content: str,
metadata: Dict = None,
importance: float = 0.5) -> str:
chunk_id = super().add_context(content, metadata, importance)
# Создаем embedding для нового контекста
embedding = self.embedding_model.encode([content])[0]
self.embeddings.append(embedding)
# Обновляем индекс
self.index.add(np.array([embedding], dtype=np.float32))
return chunk_id
def semantic_search(self,
query: str,
top_k: int = 5,
similarity_threshold: float = 0.3) -> List[ContextChunk]:
"""Семантический поиск по контексту"""
# Векторизация запроса
query_embedding = self.embedding_model.encode([query])[0]
# Поиск в индексе
distances, indices = self.index.search(
np.array([query_embedding], dtype=np.float32),
min(top_k * 2, len(self.context_chunks))
)
# Конвертируем расстояния в схожесть
similarities = 1 / (1 + distances[0])
# Фильтрация и ранжирование
results = []
for idx, similarity in zip(indices[0], similarities):
if similarity >= similarity_threshold and idx < len(self.context_chunks):
chunk = self.context_chunks[idx]
# Комбинируем семантическую схожесть с важностью
combined_score = similarity * (0.7 + 0.3 * chunk.importance)
results.append((combined_score, chunk))
# Сортировка по комбинированному score
results.sort(key=lambda x: x[0], reverse=True)
return [chunk for _, chunk in results[:top_k]]
def get_hybrid_context(self,
query: str,
top_k: int = 5,
semantic_weight: float = 0.7) -> str:
"""Гибридный поиск (семантический + ключевые слова)"""
# Семантический поиск
semantic_results = self.semantic_search(query, top_k=top_k)
# Поиск по ключевым словам (из родительского класса)
keyword_results_text = super().get_relevant_context(query, top_k=top_k)
# Комбинирование результатов
all_chunks = {}
# Добавляем семантические результаты
for chunk in semantic_results:
all_chunks[chunk.chunk_id] = (
chunk,
semantic_weight * chunk.importance
)
# Добавляем результаты по ключевым словам
for chunk in self.context_chunks:
if chunk.content in keyword_results_text:
if chunk.chunk_id in all_chunks:
# Увеличиваем score, если чанк уже есть
all_chunks[chunk.chunk_id] = (
chunk,
all_chunks[chunk.chunk_id][1] + (1 - semantic_weight) * chunk.importance
)
else:
all_chunks[chunk.chunk_id] = (
chunk,
(1 - semantic_weight) * chunk.importance
)
# Сортировка и выборка
sorted_chunks = sorted(
all_chunks.values(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return "\n\n".join([chunk.content for chunk, _ in sorted_chunks])
def save_embeddings(self, filename: str):
"""Сохранение embeddings и индекса"""
# Сохраняем базовый контекст
self.save_context(filename.replace('.pkl', '_context.json'))
# Сохраняем embeddings и индекс
data = {
'embeddings': self.embeddings,
'index': faiss.serialize_index(self.index) if self.index else None
}
with open(filename, 'wb') as f:
pickle.dump(data, f)
def load_embeddings(self, filename: str):
"""Загрузка embeddings и индекса"""
# Загружаем базовый контекст
self.load_context(filename.replace('.pkl', '_context.json'))
# Загружаем embeddings и индекс
with open(filename, 'rb') as f:
data = pickle.load(f)
self.embeddings = data['embeddings']
if data['index']:
self.index = faiss.deserialize_index(data['index'])
else:
self._build_index()
if self.embeddings:
self.index.add(np.array(self.embeddings, dtype=np.float32))
# Пример использования продвинутого MCP
def advanced_mcp_example():
"""Пример использования продвинутого MCP"""
mcp = AdvancedMCP(max_chunks=10)
# Добавляем разнообразный контекст
contexts = [
("Машинное обучение - это подраздел искусственного интеллекта.", 0.9),
("Глубокое обучение использует нейронные сети с множеством слоев.", 0.8),
("Python является популярным языком для data science.", 0.7),
("TensorFlow и PyTorch - основные фреймворки для глубокого обучения.", 0.85),
("Обучение с учителем требует размеченных данных.", 0.75),
("Рекуррентные нейронные сети хорошо работают с последовательностями.", 0.8),
("Трансформеры изменили подход к обработке естественного языка.", 0.9),
("BERT и GPT - примеры моделей-трансформеров.", 0.85),
("Обучение с подкреплением используется в робототехнике и играх.", 0.7),
("Кластеризация - пример обучения без учителя.", 0.65),
]
for content, importance in contexts:
mcp.add_context(content, importance=importance)
# Тестируем семантический поиск
query = "Какие есть методы ИИ?"
print(f"Запрос: {query}")
print("\nСемантический поиск:")
semantic_results = mcp.semantic_search(query, top_k=3)
for i, chunk in enumerate(semantic_results, 1):
print(f"{i}. {chunk.content[:100]}... (важность: {chunk.importance:.2f})")
print("\n" + "="*50 + "\n")
# Гибридный поиск
print("Гибридный поиск:")
hybrid_context = mcp.get_hybrid_context(query, top_k=4)
print(hybrid_context)
# Сохраняем состояние
mcp.save_embeddings("mcp_advanced_state.pkl")
return mcp
2.3 Реализация с графом зависимостей
import networkx as nx
from enum import Enum
class RelationType(Enum):
"""Типы отношений между чанками контекста"""
PREREQUISITE = "prerequisite" # Необходимое условие
RELATED = "related" # Связанная тема
DETAILS = "details" # Детализация
EXAMPLE = "example" # Пример
CONTRAST = "contrast" # Противопоставление
class GraphBasedMCP(AdvancedMCP):
"""MCP с графом зависимостей между чанками"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.graph = nx.DiGraph() # Ориентированный граф
def add_context_with_relations(self,
content: str,
metadata: Dict = None,
importance: float = 0.5,
relations: List[tuple] = None) -> str:
"""Добавление контекста с указанием отношений"""
chunk_id = self.add_context(content, metadata, importance)
# Добавляем узел в граф
self.graph.add_node(chunk_id,
content=content,
importance=importance,
metadata=metadata or {})
# Добавляем отношения
if relations:
for target_id, relation_type in relations:
if target_id in self.graph.nodes:
self.graph.add_edge(
target_id,
chunk_id,
relation=relation_type.value,
weight=self._calculate_relation_weight(relation_type)
)
return chunk_id
def _calculate_relation_weight(self, relation_type: RelationType) -> float:
"""Вычисление веса отношения"""
weights = {
RelationType.PREREQUISITE: 0.9,
RelationType.RELATED: 0.6,
RelationType.DETAILS: 0.7,
RelationType.EXAMPLE: 0.5,
RelationType.CONTRAST: 0.4
}
return weights.get(relation_type, 0.5)
def get_context_with_dependencies(self,
chunk_id: str,
depth: int = 2) -> List[ContextChunk]:
"""Получение контекста с зависимостями"""
if chunk_id not in self.graph.nodes:
return []
# Находим все связанные узлы до указанной глубины
related_nodes = set()
# Предшествующие узлы (предварительные знания)
predecessors = set()
for d in range(1, depth + 1):
preds = set(self.graph.predecessors(chunk_id))
predecessors.update(preds)
# Рекурсивно для каждого предшественника
for pred in list(preds):
pred_preds = set(self.graph.predecessors(pred))
predecessors.update(pred_preds)
# Последующие узлы (развитие темы)
successors = set()
for d in range(1, depth + 1):
succs = set(self.graph.successors(chunk_id))
successors.update(succs)
# Рекурсивно для каждого последователя
for succ in list(succs):
succ_succs = set(self.graph.successors(succ))
successors.update(succ_succs)
related_nodes.update(predecessors)
related_nodes.update(successors)
related_nodes.add(chunk_id)
# Получаем чанки для всех связанных узлов
result_chunks = []
for node_id in related_nodes:
for chunk in self.context_chunks:
if chunk.chunk_id == node_id:
# Увеличиваем важность для основного чанка
if node_id == chunk_id:
chunk = ContextChunk(
content=chunk.content,
metadata=chunk.metadata,
timestamp=chunk.timestamp,
importance=min(1.0, chunk.importance * 1.2),
chunk_id=chunk.chunk_id
)
result_chunks.append(chunk)
break
# Сортировка по важности
result_chunks.sort(key=lambda x: x.importance, reverse=True)
return result_chunks
def visualize_graph(self, filename: str = "context_graph.png"):
"""Визуализация графа зависимостей"""
try:
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 8))
# Позиционирование узлов
pos = nx.spring_layout(self.graph, k=2, iterations=50)
# Рисуем узлы с цветом по важности
node_colors = []
for node in self.graph.nodes():
importance = self.graph.nodes[node].get('importance', 0.5)
node_colors.append(importance)
nx.draw_networkx_nodes(
self.graph, pos,
node_color=node_colors,
node_size=500,
cmap=plt.cm.RdYlGn,
alpha=0.8
)
# Рисуем ребра с разными стилями для разных типов отношений
edge_styles = {
'prerequisite': 'solid',
'related': 'dashed',
'details': 'dotted',
'example': 'dashdot',
'contrast': (0, (3, 5, 1, 5))
}
for edge in self.graph.edges(data=True):
relation = edge[2].get('relation', 'related')
style = edge_styles.get(relation, 'solid')
nx.draw_networkx_edges(
self.graph, pos,
edgelist=[(edge[0], edge[1])],
style=style,
alpha=0.6,
width=edge[2].get('weight', 0.5) * 3
)
# Подписи узлов (первые 20 символов контента)
labels = {}
for node in self.graph.nodes():
content = self.graph.nodes[node].get('content', '')
labels[node] = f"{node[:4]}...\n{content[:20]}..."
nx.draw_networkx_labels(self.graph, pos, labels, font_size=8)
plt.title("Граф зависимостей контекста")
plt.axis('off')
plt.tight_layout()
plt.savefig(filename, dpi=300, bbox_inches='tight')
plt.close()
print(f"Граф сохранен в {filename}")
except ImportError:
print("Для визуализации установите matplotlib: pip install matplotlib")
# Пример использования графового MCP
def graph_mcp_example():
"""Пример использования MCP с графом зависимостей"""
mcp = GraphBasedMCP(max_chunks=15)
# Добавляем контекст с отношениями
# Базовые концепции
ml_id = mcp.add_context_with_relations(
"Машинное обучение (ML) - это подраздел ИИ, "
"фокусирующийся на создании систем, "
"которые обучаются на данных.",
importance=0.9
)
dl_id = mcp.add_context_with_relations(
"Глубокое обучение (Deep Learning) - подраздел ML, "
"использующий многослойные нейронные сети.",
importance=0.85,
relations=[(ml_id, RelationType.PREREQUISITE)]
)
nn_id = mcp.add_context_with_relations(
"Нейронные сети - вычислительные системы, "
"вдохновленные биологическими нейронными сетями.",
importance=0.8,
relations=[(dl_id, RelationType.DETAILS)]
)
# Примеры применения
cv_id = mcp.add_context_with_relations(
"Компьютерное зрение - применение глубокого обучения "
"для анализа изображений и видео.",
importance=0.75,
relations=[(dl_id, RelationType.EXAMPLE)]
)
nlp_id = mcp.add_context_with_relations(
"Обработка естественного языка (NLP) - применение ML "
"для работы с человеческим языком.",
importance=0.75,
relations=[(ml_id, RelationType.EXAMPLE)]
)
# Контрастные концепции
supervised_id = mcp.add_context_with_relations(
"Обучение с учителем требует размеченных данных "
"для обучения моделей.",
importance=0.7,
relations=[(ml_id, RelationType.DETAILS)]
)
unsupervised_id = mcp.add_context_with_relations(
"Обучение без учителя работает с неразмеченными данными, "
"находя скрытые паттерны.",
importance=0.7,
relations=[
(ml_id, RelationType.DETAILS),
(supervised_id, RelationType.CONTRAST)
]
)
# Получаем контекст с зависимостями
print("Контекст для глубокого обучения с зависимостями:")
related_chunks = mcp.get_context_with_dependencies(dl_id, depth=2)
for i, chunk in enumerate(related_chunks, 1):
print(f"{i}. [{chunk.importance:.2f}] {chunk.content[:80]}...")
print("\n" + "="*50 + "\n")
# Визуализируем граф
mcp.visualize_graph("mcp_graph.png")
# Семантический поиск в графе
query = "Как нейронные сети используются в ИИ?"
print(f"\nСемантический поиск для: '{query}'")
semantic_results = mcp.semantic_search(query, top_k=3)
for i, chunk in enumerate(semantic_results, 1):
print(f"{i}. {chunk.content[:80]}...")
return mcp
3. Лучшие практики и полезные советы
3.1 Оптимизация производительности
class OptimizedMCP(ModelContextProtocol):
"""Оптимизированная версия MCP"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.chunk_cache = {} # Кэш для быстрого доступа
self.access_counter = {} # Счетчик обращений
def add_context(self, *args, **kwargs):
"""Добавление с кэшированием"""
chunk_id = super().add_context(*args, **kwargs)
# Обновляем кэш
if self.context_chunks:
last_chunk = self.context_chunks[-1]
self.chunk_cache[chunk_id] = last_chunk
self.access_counter[chunk_id] = 0
return chunk_id
def get_chunk_by_id(self, chunk_id: str) -> ContextChunk:
"""Быстрый доступ к чанку по ID"""
if chunk_id in self.chunk_cache:
self.access_counter[chunk_id] += 1
return self.chunk_cache[chunk_id]
# Поиск в основном списке
for chunk in self.context_chunks:
if chunk.chunk_id == chunk_id:
self.chunk_cache[chunk_id] = chunk
self.access_counter[chunk_id] = 1
return chunk
raise ValueError(f"Чанк с ID {chunk_id} не найден")
def cleanup_cache(self, max_cache_size: int = 100):
"""Очистка кэша"""
if len(self.chunk_cache) > max_cache_size:
# Удаляем наименее используемые элементы
sorted_items = sorted(
self.access_counter.items(),
key=lambda x: x[1]
)
items_to_remove = sorted_items[:len(sorted_items) - max_cache_size]
for chunk_id, _ in items_to_remove:
if chunk_id in self.chunk_cache:
del self.chunk_cache[chunk_id]
if chunk_id in self.access_counter:
del self.access_counter[chunk_id]
3.2 Стратегии управления контекстом
class ContextManagementStrategy:
"""Стратегии управления контекстом"""
@staticmethod
def fifo_strategy(chunks: List[ContextChunk], max_chunks: int):
"""First-In-First-Out стратегия"""
return chunks[-max_chunks:] if len(chunks) > max_chunks else chunks
@staticmethod
def importance_based_strategy(chunks: List[ContextChunk], max_chunks: int):
"""Стратегия на основе важности"""
sorted_chunks = sorted(chunks, key=lambda x: x.importance, reverse=True)
return sorted_chunks[:max_chunks]
@staticmethod
def recency_based_strategy(chunks: List[ContextChunk], max_chunks: int):
"""Стратегия на основе новизны"""
sorted_chunks = sorted(chunks, key=lambda x: x.timestamp, reverse=True)
return sorted_chunks[:max_chunks]
@staticmethod
def hybrid_strategy(chunks: List[ContextChunk], max_chunks: int,
importance_weight: float = 0.6,
recency_weight: float = 0.4):
"""Гибридная стратегия"""
if not chunks:
return []
# Нормализуем важность и время
max_importance = max(chunk.importance for chunk in chunks)
newest_time = max(chunk.timestamp for chunk in chunks)
oldest_time = min(chunk.timestamp for chunk in chunks)
time_range = (newest_time - oldest_time).total_seconds()
scored_chunks = []
for chunk in chunks:
# Нормализованная важность
norm_importance = chunk.importance / max_importance if max_importance > 0 else 0
# Нормализованная новизна
if time_range > 0:
recency = (chunk.timestamp - oldest_time).total_seconds() / time_range
else:
recency = 1.0
# Комбинированный score
score = (importance_weight * norm_importance +
recency_weight * recency)
scored_chunks.append((score, chunk))
# Сортировка по score
scored_chunks.sort(key=lambda x: x[0], reverse=True)
return [chunk for _, chunk in scored_chunks[:max_chunks]]
class StrategicMCP(ModelContextProtocol):
"""MCP с поддержкой разных стратегий управления"""
def __init__(self, strategy: str = "hybrid", *args, **kwargs):
super().__init__(*args, **kwargs)
self.strategy = strategy
self.strategies = {
'fifo': ContextManagementStrategy.fifo_strategy,
'importance': ContextManagementStrategy.importance_based_strategy,
'recency': ContextManagementStrategy.recency_based_strategy,
'hybrid': ContextManagementStrategy.hybrid_strategy
}
def _manage_context_size(self):
"""Управление контекстом с выбранной стратегией"""
if self.strategy in self.strategies:
self.context_chunks = self.strategies[self.strategy](
self.context_chunks,
self.max_chunks
)
else:
# Используем стратегию по умолчанию
super()._manage_context_size()
3.3 Интеграция с популярными LLM
class LLMIntegrationMCP(ModelContextProtocol):
"""MCP с интеграцией популярных LLM"""
def __init__(self, llm_provider: str = "openai", *args, **kwargs):
super().__init__(*args, **kwargs)
self.llm_provider = llm_provider
def generate_with_context(self,
prompt: str,
model: str = None,
temperature: float = 0.7) -> str:
"""Генерация ответа с использованием контекста"""
# Получаем релевантный контекст
context = self.get_relevant_context(prompt)
# Формируем промпт с контекстом
full_prompt = self._format_prompt(prompt, context)
# Генерация ответа (заглушка - в реальности используйте API LLM)
response = self._call_llm_api(full_prompt, model, temperature)
# Обновляем контекст с ответом
self.add_context(
f"Ответ на: {prompt[:50]}...\n{response[:100]}...",
importance=0.6
)
return response
def _format_prompt(self, prompt: str, context: str) -> str:
"""Форматирование промпта с контекстом"""
prompt_templates = {
"openai": f"""На основе следующего контекста ответь на вопрос.
Контекст:
{context}
Вопрос: {prompt}
Ответ:""",
"anthropic": f"""Human: Вот контекст для справки:
<context>
{context}
</context>
Теперь ответь на вопрос: {prompt}
Assistant:""",
"cohere": f"""Контекст: {context}
Вопрос: {prompt}
Ответ:"""
}
return prompt_templates.get(
self.llm_provider,
f"Контекст: {context}\n\nВопрос: {prompt}\n\nОтвет:"
)
def _call_llm_api(self,
prompt: str,
model: str = None,
temperature: float = 0.7) -> str:
"""Вызов LLM API (заглушка для примера)"""
# В реальной реализации здесь будет интеграция с API
# OpenAI, Anthropic, Google, etc.
# Заглушка
llm_responses = {
"openai": "Это ответ, сгенерированный на основе предоставленного контекста.",
"anthropic": "Основываясь на контексте, я могу сказать, что...",
"cohere": "Согласно контексту, ответ следующий..."
}
return llm_responses.get(
self.llm_provider,
"Ответ сгенерирован с использованием контекста."
)
4. Практические рекомендации
4.1 Когда использовать разные стратегии
def choose_strategy_based_on_use_case():
"""Выбор стратегии управления контекстом в зависимости от задачи"""
strategies_guide = {
"диалоговая система": {
"рекомендация": "hybrid",
"обоснование": "Баланс между важностью и новизной сообщений",
"параметры": {"importance_weight": 0.4, "recency_weight": 0.6}
},
"техническая документация": {
"рекомендация": "importance",
"обоснование": "Приоритет ключевых концепций и определений",
"параметры": {"max_chunks": 20}
},
"новостной агрегатор": {
"рекомендация": "recency",
"обоснование": "Актуальность информации критически важна",
"параметры": {"max_chunks": 15}
},
"система поддержки": {
"рекомендация": "fifo",
"обоснование": "Последовательность диалога важнее релевантности",
"параметры": {"max_chunks": 30}
}
}
return strategies_guide
4.2 Мониторинг и анализ
class MonitoringMCP(ModelContextProtocol):
"""MCP с мониторингом и аналитикой"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.metrics = {
'total_chunks_added': 0,
'chunks_evicted': 0,
'cache_hits': 0,
'cache_misses': 0,
'avg_importance': 0.0,
'response_times': []
}
def add_context(self, *args, **kwargs):
"""Добавление с отслеживанием метрик"""
import time
start_time = time.time()
result = super().add_context(*args, **kwargs)
end_time = time.time()
self.metrics['response_times'].append(end_time - start_time)
self.metrics['total_chunks_added'] += 1
# Обновляем среднюю важность
total_importance = sum(chunk.importance for chunk in self.context_chunks)
self.metrics['avg_importance'] = (
total_importance / len(self.context_chunks)
if self.context_chunks else 0
)
return result
def get_performance_report(self) -> Dict:
"""Отчет о производительности"""
import statistics
report = {
'current_chunks': len(self.context_chunks),
'total_chunks_added': self.metrics['total_chunks_added'],
'chunks_evicted': self.metrics['chunks_evicted'],
'cache_hit_rate': (
self.metrics['cache_hits'] /
(self.metrics['cache_hits'] + self.metrics['cache_misses'])
if (self.metrics['cache_hits'] + self.metrics['cache_misses']) > 0
else 0
),
'average_importance': self.metrics['avg_importance'],
'avg_response_time': (
statistics.mean(self.metrics['response_times'][-100:])
if self.metrics['response_times'] else 0
),
'context_utilization': (
len(self.context_chunks) / self.max_chunks
if self.max_chunks > 0 else 0
)
}
return report
def visualize_metrics(self):
"""Визуализация метрик"""
try:
import matplotlib.pyplot as plt
import numpy as np
report = self.get_performance_report()
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
# 1. Использование контекста
labels = ['Используется', 'Свободно']
sizes = [
report['context_utilization'] * 100,
(1 - report['context_utilization']) * 100
]
axes[0, 0].pie(sizes, labels=labels, autopct='%1.1f%%')
axes[0, 0].set_title('Использование контекста')
# 2. Важность чанков
importances = [chunk.importance for chunk in self.context_chunks]
if importances:
axes[0, 1].hist(importances, bins=10, alpha=0.7)
axes[0, 1].set_xlabel('Важность')
axes[0, 1].set_ylabel('Количество')
axes[0, 1].set_title('Распределение важности чанков')
# 3. Время ответа
if self.metrics['response_times']:
axes[1, 0].plot(
range(len(self.metrics['response_times'][-50:])),
self.metrics['response_times'][-50:],
marker='o'
)
axes[1, 0].set_xlabel('Запросы')
axes[1, 0].set_ylabel('Время (сек)')
axes[1, 0].set_title('Время ответа')
axes[1, 0].grid(True)
# 4. Хитрейт кэша
cache_data = {
'Hits': report['cache_hit_rate'] * 100,
'Misses': (1 - report['cache_hit_rate']) * 100
}
axes[1, 1].bar(cache_data.keys(), cache_data.values())
axes[1, 1].set_ylabel('Процент')
axes[1, 1].set_title('Эффективность кэша')
plt.tight_layout()
plt.savefig('mcp_metrics.png', dpi=300)
plt.close()
print("Метрики сохранены в mcp_metrics.png")
except ImportError:
print("Для визуализации установите matplotlib")
5. Заключение
5.1 Ключевые выводы
- MCP необходим для эффективной работы с ограниченным контекстом LLM
- Разные стратегии подходят для разных use-case
- Граф зависимостей улучшает связность контекста
- Мониторинг метрик помогает оптимизировать производительность
- Кэширование и индексация критически важны для скорости
5.2 Рекомендации по внедрению
def implementation_checklist():
"""Чеклист внедрения MCP"""
checklist = {
"1. Анализ требований": [
"Определить максимальный размер контекста",
"Определить типы информации для хранения",
"Определить критерии релевантности"
],
"2. Выбор стратегии": [
"Выбрать стратегию управления контекстом",
"Определить веса для гибридных стратегий",
"Настроить параметры устаревания"
],
"3. Реализация": [
"Имплементировать базовый MCP",
"Добавить семантический поиск при необходимости",
"Реализовать граф зависимостей для сложных сценариев"
],
"4. Оптимизация": [
"Добавить кэширование",
"Реализовать индексацию для быстрого поиска",
"Настроить мониторинг производительности"
],
"5. Тестирование": [
"Тестировать на различных наборах данных",
"Измерять точность поиска",
"Оценивать производительность под нагрузкой"
]
}
return checklist
5.3 Дальнейшее развитие
Для продвинутого использования рассмотрите:
- Интеграция с векторными БД (Pinecone, Weaviate, Qdrant)
- Использование Transformer моделей для оценки релевантности
- Мультимодальный контекст (текст + изображения + аудио)
- Распределенный MCP для масштабирования
- Адаптивное обучение стратегий управления контекстом