使用模型上下文協議有效地管理上下文

在本教程中，我們通過構建ModelContextManager來指導您實現模型上下文協議（MCP）

```python

``python

import torch

導入火炬

import numpy as np

導入numpy作為NP

import typing

導入打字

from dataclasses import dataclass

從數據級導入數據級

import time

進口時間

import gc

導入GC

from tqdm.notebook import tqdm

來自tqdm.notebook導入tqdm

from sentence_transformers import SentenceTransformer

從stone_transformers導入句子詞術語

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'

設備='cuda'如果torch.cuda.is_available（）else'cpu'

NUM_CHUNKS = 50

num_chunks = 50

CHUNK_SIZE = 100

chunk_size = 100

RELEVANCE_THRESHOLD = 0.1

相關性_threshold = 0.1

IMPORTANCE_FACTOR = 1.0

Questional_factor = 1.0

RECENCY_FACTOR = 0.5

recency_factor = 0.5

VISUALIZE_CONTEXT = True

Visualize_Context = true

BATCH_SIZE = 32

batch_size = 32

class ContextChunk(typing.NamedTuple):

班級上下文chunk（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 =無

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，gpt2_tokenizer：gpt2_tokenizer：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_pretrataining（“ gpt2”）

else:

別的：

self.gpt2_tokenizer = gpt2_tokenizer

self.gpt2_tokenizer = gpt2_tokenizer

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

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

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

def add_chunk（self，chunk_text：str，重要性：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_embedding = 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，重要性=重要性）

self.context_chunks.append(new_chunk)

self.context_chunks.sappend（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]）}令牌和重要性{quiental}。

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

def optimize_context_window（自我，查詢：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_embedding = self.sentence_transformer.encode（query，batch_size = batch_size）

chunks_to_keep = []

chunks_to_keep = []

remaining_tokens = self.max_tokens - self.used_tokens

剩餘_tokens = self.max_tokens- self.used_tokens

if remaining_tokens < 0:

如果剩下的_tokens <0：

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

打印（“警告：令牌限制超過％s令牌”％ - remaining_tokens）

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

對於我的範圍（min_chunks，len（self.context_chunks）-1，-1）：

chunk = self.context_chunks[i]

塊= 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_recency = (time.time() - chunk.timestamp) * RECENCY_FACTOR

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

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

相關_scores = np.Array（[cosine_simurility（chunk.embedding，x）for query_embedding中的x]）

max_relevant_score = np.max(relevant_scores)

max_relevant_score = np.max（相關_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> = sensepance_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:

如果剩下的_tokens> = chunk_token_count：

chunks_to_keep.append(i)

chunks_to_keep.append（i）

remaining_

其餘的_