モデルコンテキストプロトコルでコンテキストを効果的に管理します

このチュートリアルでは、ModelContextManagerを構築することにより、モデルコンテキストプロトコル（MCP）の実用的な実装をご案内します

```python

`` python

import torch

トーチをインポートします

import numpy as np

npとしてnumpyをインポートします

import typing

インポートタイピング

from dataclasses import dataclass

DataClassesからImport Dataclassから

import time

インポート時間

import gc

GCをインポートします

from tqdm.notebook import tqdm

tqdm.notebookからインポートtqdmから

from sentence_transformers import SentenceTransformer

SentenceTransFormerをImportからexcentから

from transformers import GPT2Tokenizer, FLAN_T5ForConditionalGeneration, AutoTokenizer, AutoModelForSeq2SeqLM

トランスから輸入gpt2tokenizer、flan_t5forconditionalgeneration、autotokenizer、automodelforseq2seqlm

import math

数学をインポートします

MAX_TOKENS = 8000

max_tokens = 8000

DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'

device = 'cuda' if torch.cuda.is_available（）else 'cpu'

NUM_CHUNKS = 50

num_chunks = 50

CHUNK_SIZE = 100

chunk_size = 100

RELEVANCE_THRESHOLD = 0.1

revervance_threshold = 0.1

IMPORTANCE_FACTOR = 1.0

compettion_factor = 1.0

RECENCY_FACTOR = 0.5

Remency_Factor = 0.5

VISUALIZE_CONTEXT = True

visualize_context = true

BATCH_SIZE = 32

batch_size = 32

class ContextChunk(typing.NamedTuple):

クラスコンテキストチャンク（typing.namedtuple）：

content: str

コンテンツ：str

embedding: np.array

埋め込み：np.array

importance: float = 1.0

重要性：float = 1.0

timestamp: float = time.time()

タイムスタンプ：float = time.time（）

metadata: dict = None

メタデータ：dict = none

def __post_init__(self):

def __post_init __（self）：

if self.metadata is None:

self.metadataがいない場合：

self.metadata = {}

self.metadata = {}

class ModelContextManager:

クラスModelContextManager：

def __init__(self, context_chunks:typing.List[ContextChunk]=None, max_tokens:int=MAX_TOKENS, token_limit:int=0, gpt2_tokenizer:GPT2Tokenizer=None):

def __ init __（self、context_chunks：typing.list [contextchunk] = none、max_tokens：int = max_tokens、token_limit：int = 0、gpt2tokenizer：gpt2tokenizer = none）：

self.max_tokens = max_tokens

self.max_tokens = max_tokens

self.token_limit = token_limit

self.token_limit = token_limit

self.context_chunks = context_chunks or []

self.context_chunks = context_chunksまたは[]

self.used_tokens = 0

self.used_tokens = 0

self.last_chunk_index = 0

self.last_chunk_index = 0

self.total_chunks = 0

self.total_chunks = 0

if gpt2_tokenizer is None:

GPT2_TOKENIZERがなしである場合：

self.gpt2_tokenizer = GPT2Tokenizer.from_pretrained("gpt2")

self.gpt2_tokenizer = gpt2tokenizer.from_pretrained（ "gpt2"）

else:

それ以外：

self.gpt2_tokenizer = gpt2_tokenizer

self.gpt2_tokenizer = gpt2_tokenizer

self.sentence_transformer = SentenceTransformer('all-mpnet-base-v2')

self.sentence_transformer = senecetransformer（ 'all-mpnet-base-v2'）

def add_chunk(self, chunk_text:str, importance:float=1.0):

def add_chunk（self、chunk_text：str、compation：float = 1.0）：

encoded_input = self.gpt2_tokenizer(chunk_text, return_tensors='pt')

encoded_input = self.gpt2_tokenizer（chunk_text、return_tensors = 'pt'）

self.used_tokens += int(encoded_input[0].shape[1])

self.used_tokens += int（encoded_input [0] .shape [1]）

chunk_embedding = self.sentence_transformer.encode(chunk_text, batch_size=BATCH_SIZE)

chunk_embeding = self.sentence_transformer.encode（chunk_text、batch_size = batch_size）

new_chunk = ContextChunk(content=chunk_text, embedding=chunk_embedding, importance=importance)

new_chunk = contextchunk（content = chunk_text、embedding = chunk_embedding、compation = compation）

self.context_chunks.append(new_chunk)

self.context_chunks.append（new_chunk）

self.last_chunk_index += 1

self.last_chunk_index += 1

self.total_chunks += 1

self.total_chunks += 1

print(f"Added chunk with {int(encoded_input[0].shape[1])} tokens and importance {importance}. Total used tokens: {self.used_tokens}, total chunks: {self.total_chunks}")

print（f "{int（encoded_input [0] .shape [1]）} tokens and compation {compettion}。

def optimize_context_window(self, query:str, min_chunks:int=3):

def optimize_context_window（self、query：str、min_chunks：int = 3）：

if len(self.context_chunks) <= min_chunks:

len（self.context_chunks）<= min_chunks：

return []

戻る []

query_embedding = self.sentence_transformer.encode(query, batch_size=BATCH_SIZE)

query_embeding = self.sentence_transformer.encode（query、batch_size = batch_size）

chunks_to_keep = []

chunks_to_keep = []

remaining_tokens = self.max_tokens - self.used_tokens

resight_tokens = self.max_tokens -self.used_tokens

if remaining_tokens < 0:

rester_tokens <0の場合：

print("Warning: token limit exceeded by %s tokens" % -remaining_tokens)

print（ "警告：トークン制限は％sトークンを超えて"％-remaining_tokens）

for i in range(min_chunks, len(self.context_chunks) - 1, -1):

範囲のi（min_chunks、len（self.context_chunks）-1、-1）：

chunk = self.context_chunks[i]

chunk = self.context_chunks [i]

if i == len(self.context_chunks) - 1:

i == len（self.context_chunks）-1：

chunks_to_keep.append(i)

chunks_to_keep.append（i）

continue

続く

chunk_importance = chunk.importance * IMPORTANCE_FACTOR

chunk_importance = chunk.importance * custition_factor

chunk_recency = (time.time() - chunk.timestamp) * RECENCY_FACTOR

chunk_recency =（time.time（） - chunk.timestamp） * Remency_Factor

relevant_scores = np.array([cosine_similarity(chunk.embedding, x) for x in query_embedding])

relevant_scores = np.array（[cosine_similarity（chunk.embeding、x）for x in query_embeding]）

max_relevant_score = np.max(relevant_scores)

MAX_RELEVANT_SCORE = np.max（relevant_scores）

total_score = chunk_importance + chunk_recency + max_relevant_score

Total_score = chunk_importance + chunk_recency + max_relevant_score

if total_score >= RELEVANCE_THRESHOLD:

Total_score> = relevance_threshold：

encoded_input = self.gpt2_tokenizer(chunk.content, return_tensors='pt')

encoded_input = self.gpt2_tokenizer（chunk.content、return_tensors = 'pt'）

chunk_token_count = int(encoded_input[0].shape[1])

chunk_token_count = int（encoded_input [0] .shape [1]）

if remaining_tokens >= chunk_token_count:

rester_tokens> = chunk_token_count：

chunks_to_keep.append(i)

chunks_to_keep.append（i）

remaining_

残り_