إدارة نافذة السياق واستراتيجيات السياق الطويل

نافذة السياق هي الحد الأقصى لعدد التوكنات التي يمكن لنموذج لغوي كبير معالجتها دفعة واحدة. الإدارة الفعّالة للسياق تؤثر بشكل مباشر على أداء تطبيقات الذكاء الاصطناعي.

حدود نافذة السياق

مقارنة النماذج

حساب التوكنات

1import tiktoken
2
3def count_tokens(text: str, model: str = "gpt-4") -> int:
4    encoding = tiktoken.encoding_for_model(model)
5    return len(encoding.encode(text))
6
7def estimate_tokens(text: str) -> int:
8    # Quick estimate: ~4 chars = 1 token (English)
9    return len(text) // 4

استراتيجيات تقسيم المحتوى (Chunking)

التقسيم بحجم ثابت

1def fixed_size_chunk(text: str, chunk_size: int = 1000, overlap: int = 200) -> list:
2    chunks = []
3    start = 0
4    
5    while start < len(text):
6        end = start + chunk_size
7        chunk = text[start:end]
8        chunks.append(chunk)
9        start = end - overlap
10    
11    return chunks

التقسيم الدلالي

1from langchain.text_splitter import RecursiveCharacterTextSplitter
2
3def semantic_chunk(text: str, chunk_size: int = 1000) -> list:
4    splitter = RecursiveCharacterTextSplitter(
5        chunk_size=chunk_size,
6        chunk_overlap=200,
7        separators=["\n\n", "\n", ". ", " ", ""],
8        length_function=len
9    )
10    
11    return splitter.split_text(text)

التقسيم المعتمد على هيكل الوثيقة

1def structure_aware_chunk(document: str) -> list:
2    chunks = []
3    current_section = ""
4    current_header = ""
5    
6    for line in document.split("\n"):
7        # Header detection
8        if line.startswith("#"):
9            if current_section:
10                chunks.append({
11                    "header": current_header,
12                    "content": current_section.strip()
13                })
14            current_header = line
15            current_section = ""
16        else:
17            current_section += line + "\n"
18    
19    if current_section:
20        chunks.append({
21            "header": current_header,
22            "content": current_section.strip()
23        })
24    
25    return chunks

ضغط السياق

التلخيص

1def compress_context(text: str, max_tokens: int = 2000) -> str:
2    current_tokens = count_tokens(text)
3    
4    if current_tokens <= max_tokens:
5        return text
6    
7    # Summarize with LLM
8    response = client.chat.completions.create(
9        model="gpt-4-turbo",
10        messages=[
11            {
12                "role": "system",
13                "content": f"Summarize the following text under {max_tokens} tokens. "
14                          "Preserve important information."
15            },
16            {"role": "user", "content": text}
17        ]
18    )
19    
20    return response.choices[0].message.content

الضغط الاستخراجي

1from sklearn.feature_extraction.text import TfidfVectorizer
2import numpy as np
3
4def extractive_compress(text: str, ratio: float = 0.3) -> str:
5    sentences = text.split(". ")
6    
7    # Find important sentences with TF-IDF
8    vectorizer = TfidfVectorizer()
9    tfidf_matrix = vectorizer.fit_transform(sentences)
10    
11    # Importance score of each sentence
12    scores = np.array(tfidf_matrix.sum(axis=1)).flatten()
13    
14    # Select most important sentences
15    num_sentences = max(1, int(len(sentences) * ratio))
16    top_indices = np.argsort(scores)[-num_sentences:]
17    top_indices = sorted(top_indices)  # Preserve order
18    
19    return ". ".join([sentences[i] for i in top_indices])
20## نافذة منزلقة
21
22### إدارة سجل المحادثة
23
24```python
25class SlidingWindowMemory:
26    def __init__(self, max_tokens: int = 4000):
27        self.max_tokens = max_tokens
28        self.messages = []
29    
30    def add_message(self, role: str, content: str):
31        self.messages.append({"role": role, "content": content})
32        self._trim()
33    
34    def _trim(self):
35        while self._total_tokens() > self.max_tokens and len(self.messages) > 2:
36            # Preserve System message, delete oldest user/assistant
37            if self.messages[0]["role"] == "system":
38                self.messages.pop(1)
39            else:
40                self.messages.pop(0)
41    
42    def _total_tokens(self) -> int:
43        return sum(count_tokens(m["content"]) for m in self.messages)
44    
45    def get_messages(self) -> list:
46        return self.messages.copy()

نافذة معالجة المستندات

1def process_long_document(document: str, query: str, window_size: int =8000):
2    chunks = semantic_chunk(document, chunk_size=window_size)
3    results = []
4    
5    for i, chunk in enumerate(chunks):
6        response = client.chat.completions.create(
7            model="gpt-4-turbo",
8            messages=[
9                {
10                    "role": "system",
11                    "content": "Analyze the given text chunk."
12                },
13                {
14                    "role": "user",
15                    "content": f"Text:\n{chunk}\n\nQuestion: {query}"
16                }
17            ]
18        )
19        
20        results.append({
21            "chunk_index": i,
22            "response": response.choices[0].message.content
23        })
24    
25    # Combine results
26    return synthesize_results(results, query)

نمط Map-Reduce

أسئلة وأجوبة المستندات الطويلة

1def map_reduce_qa(document: str, question: str):
2    chunks = semantic_chunk(document, chunk_size=4000)
3    
4    # Map: Analyze each chunk separately
5    partial_answers = []
6    for chunk in chunks:
7        response = client.chat.completions.create(
8            model="gpt-4-turbo",
9            messages=[
10                {
11                    "role": "user",
12                    "content": f"Text:\n{chunk}\n\nQuestion: {question}\n\n"
13                              "Answer based on this text chunk. "
14                              "If no information, say 'No information in this chunk'."
15                }
16            ]
17        )
18        partial_answers.append(response.choices[0].message.content)
19    
20    # Reduce: Combine answers
21    combined = "\n\n".join([
22        f"Source {i+1}: {ans}" 
23        for i, ans in enumerate(partial_answers)
24    ])
25    
26    final_response = client.chat.completions.create(
27        model="gpt-4-turbo",
28        messages=[
29            {
30                "role": "user",
31                "content": f"Information from different sources:\n{combined}\n\n"
32                          f"Question: {question}\n\n"
33                          "Provide a comprehensive answer by synthesizing all information."
34            }
35        ]
36    )
37    
38    return final_response.choices[0].message.content
39## تعزيز الاسترجاع داخل السياق
40
41### اختيار السياق بذكاء
42
43```python
44def select_relevant_context(query: str, documents: list, max_tokens: int = 4000):
45    # Embedding-based relevance
46    query_embedding = get_embedding(query)
47    
48    scored_docs = []
49    for doc in documents:
50        doc_embedding = get_embedding(doc["content"])
51        score = cosine_similarity(query_embedding, doc_embedding)
52        scored_docs.append({"doc": doc, "score": score})
53    
54    # Sort by relevance
55    scored_docs.sort(key=lambda x: x["score"], reverse=True)
56    
57    # Add until Token limit
58    selected = []
59    current_tokens = 0
60    
61    for item in scored_docs:
62        doc_tokens = count_tokens(item["doc"]["content"])
63        if current_tokens + doc_tokens <= max_tokens:
64            selected.append(item["doc"])
65            current_tokens += doc_tokens
66        else:
67            break
68    
69    return selected

أفضل الممارسات للسياق الطويل

1. تموضع الـ Prompt

1def optimize_prompt_position(context: str, query: str) -> str:
2    """Put important information at start and end (Lost in the Middle)"""
3    
4    chunks = semantic_chunk(context)
5    
6    # Preserve first and last chunks
7    if len(chunks) > 2:
8        middle = chunks[1:-1]
9        compressed_middle = compress_context(" ".join(middle))
10        context = f"{chunks[0]}\n\n{compressed_middle}\n\n{chunks[-1]}"
11    
12    return f"Context:\n{context}\n\n---\n\nQuestion: {query}"

2. المعالجة الهرمية

1def hierarchical_summarize(document: str, levels: int = 2):
2    """Hierarchical summarization"""
3    
4    current_text = document
5    
6    for level in range(levels):
7        chunks = semantic_chunk(current_text, chunk_size=4000)
8        
9        summaries = []
10        for chunk in chunks:
11            summary = compress_context(chunk, max_tokens=500)
12            summaries.append(summary)
13        
14        current_text = "\n\n".join(summaries)
15    
16    return current_text

3. مرساة الانتباه

1def add_attention_anchors(prompt: str) -> str:
2    """Add attention anchors"""
3    
4    return f"""
5[IMPORTANT START]
6{prompt[:500]}
7[/IMPORTANT]
8
9{prompt[500:-500]}
10
11[IMPORTANT END]
12{prompt[-500:]}
13[/IMPORTANT]
14"""

المراقبة واستكشاف الأخطاء

1class ContextMonitor:
2    def __init__(self):
3        self.logs = []
4    
5    def log_request(self, messages: list, model: str):
6        total_tokens = sum(count_tokens(m["content"]) for m in messages)
7        
8        self.logs.append({
9            "timestamp": datetime.now(),
10            "model": model,
11            "input_tokens": total_tokens,
12            "message_count": len(messages)
13        })
14        
15        # Alerts
16        if total_tokens > 100000:
17            print(f"⚠️ High token count: {total_tokens}")
18    
19    def get_stats(self):
20        return {
21            "avg_tokens": np.mean([l["input_tokens"] for l in self.logs]),
22            "max_tokens": max(l["input_tokens"] for l in self.logs),
23            "total_requests": len(self.logs)
24        }

الخلاصة

إدارة نافذة السياق عامل أساسي في قابلية التوسع وتكاليف تطبيقات LLM. يمكنك العمل بفعالية مع المستندات الطويلة باستخدام التجزئة والضغط واستراتيجيات الاسترجاع الذكي.

في Veni AI، نقوم بتطوير حلول ذكاء اصطناعي مخصصة للسياقات الطويلة.