上下文工程：让 AI 拥有正确的知识

1
2
3
4

知识层设计原则:
  - 最小化：避免不必要的冗余信息
  - 结构化：使用明确的格式便于解析
  - 层次化：核心知识优先，专业知识按需加载

1
2
3
4
5
6
7
8

# 优秀的系统提示设计
system_prompt = """
你是一个专业的软件开发助手，专注于：
1. 代码质量：可读性、可维护性、性能
2. 最佳实践：设计模式、架构原则、编码规范  
3. 安全考虑：输入验证、错误处理、权限控制
请使用中文回复，专业术语保留英文。
"""

1
2
3
4
5

Token 1 → Key1, Value1
Token 2 → Key2, Value2  
Token 3 → Key3, Value3
...
Token N → KeyN, ValueN

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 智能记忆压缩
def compress_memory(memory_chunks, max_tokens=4000):
    """
    对记忆进行重要性排序和压缩
    """
    # 1. 按重要性排序
    ranked_chunks = rank_by_importance(memory_chunks)
    
    # 2. 渐进式压缩
    compressed = []
    current_tokens = 0
    
    for chunk in ranked_chunks:
        if current_tokens + chunk['tokens'] <= max_tokens:
            compressed.append(chunk)
            current_tokens += chunk['tokens']
        else:
            break
    
    return compressed

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from sentence_transformers import SentenceTransformer

# 加载预训练模型
model = SentenceTransformer('all-MiniLM-L6-v2')

# 文档向量化
documents = [
    "Python 是一种解释型编程语言",
    "机器学习是人工智能的分支", 
    "深度学习使用神经网络架构"
]

embeddings = model.encode(documents)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def smart_chunking(text, max_length=512, overlap=50):
    """
    智能分块算法：
    1. 按段落自然分割
    2. 保持语义完整性
    3. 添加必要的上下文标记
    """
    chunks = []
    paragraphs = text.split('\n\n')
    
    for para in paragraphs:
        if len(para) <= max_length:
            chunks.append(para)
        else:
            # 按句子进一步分割
            sentences = split_into_sentences(para)
            current_chunk = ""
            
            for sentence in sentences:
                if len(current_chunk) + len(sentence) <= max_length:
                    current_chunk += sentence + " "
                else:
                    chunks.append(current_chunk.strip())
                    current_chunk = sentence + " "
            
            if current_chunk:
                chunks.append(current_chunk.strip())
    
    return chunks

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from rank_bm25 import BM25Okapi

def lost_in_middle_fix(documents, query):
    """
    解决"Lost in the Middle"问题的重排序算法
    """
    # 1. BM25 初步排序
    tokenized_docs = [doc.split() for doc in documents]
    bm25 = BM25Okapi(tokenized_docs)
    
    # 2. 位置权重调整
    doc_scores = []
    for i, doc in enumerate(documents):
        # 给开头和结尾更高的权重
        position_weight = 1.0
        if i < len(documents) * 0.2 or i > len(documents) * 0.8:
            position_weight = 1.5
        
        score = bm25.get_scores(doc.split(), query)[0] * position_weight
        doc_scores.append((i, score))
    
    # 3. 重新排序
    doc_scores.sort(key=lambda x: x[1], reverse=True)
    return [documents[i] for i, _ in doc_scores]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

def structured_generation(prompt, schema):
    """
    结构化生成控制器
    """
    # 1. 添加格式约束
    constrained_prompt = f"""
请严格按照以下格式回答：

{schema['format']}

用户问题：{prompt}

请确保：
- 使用指定的格式
- 内容准确完整
- 语言清晰易懂
"""
    
    # 2. 调用模型
    response = call_llm(constrained_prompt)
    
    # 3. 格式验证和修正
    if validate_format(response, schema):
        return response
    else:
        return fix_format(response, schema)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

def value_based_allocation(context_tokens, total_budget):
    """
    基于价值的预算分配策略
    """
    priorities = {
        'system_prompt': 0.15,      # 系统指令 15%
        'core_context': 0.25,       # 核心上下文 25%
        'recent_history': 0.20,     # 近期历史 20%
        'tool_definitions': 0.15,   # 工具定义 15%
        'retrieved_docs': 0.25      # 检索文档 25%
    }
    
    allocated = {}
    remaining = total_budget
    
    for component, ratio in priorities.items():
        component_budget = int(total_budget * ratio)
        allocated[component] = min(component_budget, remaining)
        remaining -= allocated[component]
    
    return allocated

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class PromptCache:
    def __init__(self):
        self.cache = {}
    
    def get_cached_prompt(self, prompt_hash):
        return self.cache.get(prompt_hash)
    
    def cache_prompt(self, prompt_hash, prompt_data):
        self.cache[prompt_hash] = {
            'data': prompt_data,
            'timestamp': time.time(),
            'usage_count': 0
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

hybrid_context_strategy = {
    'short_term': {
        'memory_type': 'working_memory',
        'max_tokens': 2000,
        'refresh_rate': 'turn_by_turn'
    },
    'long_term': {
        'memory_type': 'vector_db', 
        'max_tokens': 8000,
        'refresh_rate': 'hourly'
    },
    'retrieval': {
        'method': 'semantic_search',
        'top_k': 5,
        'reranking': True
    }
}

1
2
3
4
5
6
7

# requirements.txt
sentence-transformers==2.2.2
chromadb==0.4.18
langchain==0.1.0
openai==1.3.7
numpy==1.24.3
pandas==2.0.3

1

pip install -r requirements.txt

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from document_processor import DocumentProcessor

# 初始化处理器
processor = DocumentProcessor(
    chunk_size=512,
    chunk_overlap=50,
    embedding_model='all-MiniLM-L6-v2'
)

# 加载文档
documents = processor.load_documents([
    'data/company_handbook.pdf',
    'data/tech_specs.md',
    'data/policies.txt'
])

# 处理文档
processed_chunks = processor.process_documents(documents)

# 创建向量数据库
vector_db = processor.create_vector_db(processed_chunks)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

class DocumentRetriever:
    def __init__(self, vector_db, reranker=True):
        self.vector_db = vector_db
        self.reranker = reranker
        self.bm25 = None
        
        if reranker:
            self.initialize_bm25()
    
    def initialize_bm25(self):
        """初始化 BM25 重排序器"""
        from rank_bm25 import BM25Okapi
        
        # 准备 BM25 索引
        all_chunks = [chunk['text'] for chunk in self.vector_db.get_all_chunks()]
        tokenized_chunks = [doc.split() for doc in all_chunks]
        self.bm25 = BM25Okapi(tokenized_chunks)
    
    def retrieve(self, query, top_k=5):
        """两阶段检索：向量搜索 + 重排序"""
        # 第一阶段：向量搜索
        vector_results = self.vector_db.search(query, top_k * 2)
        
        # 第二阶段：重排序
        if self.reranker and self.bm25:
            reranked = self.rerank_results(query, vector_results)
            return reranked[:top_k]
        else:
            return vector_results[:top_k]
    
    def rerank_results(self, query, results):
        """处理 Lost in the Middle 问题"""
        texts = [r['text'] for r in results]
        scores = self.bm25.get_scores(query.split(), texts)
        
        # 添加位置权重
        final_scores = []
        for i, (result, score) in enumerate(zip(results, scores)):
            position_weight = 1.0
            if i < len(results) * 0.2 or i > len(results) * 0.8:
                position_weight = 1.3
            
            final_scores.append((result, score * position_weight))
        
        # 重新排序
        final_scores.sort(key=lambda x: x[1], reverse=True)
        return [result for result, _ in final_scores]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

class RAGSystem:
    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm_client = llm_client
    
    def generate_response(self, query, context_window=4000):
        """生成回答"""
        # 1. 检索相关文档
        relevant_docs = self.retriever.retrieve(query)
        
        # 2. 上下文构建
        context = self.build_context(relevant_docs, context_window)
        
        # 3. 生成回答
        prompt = self.create_prompt(query, context)
        response = self.llm_client.generate(prompt)
        
        return response, relevant_docs
    
    def build_context(self, docs, max_tokens):
        """智能上下文构建"""
        context_parts = []
        used_tokens = 0
        
        # 按重要性排序文档
        sorted_docs = self.rank_documents_by_importance(docs)
        
        for doc in sorted_docs:
            if used_tokens + doc['tokens'] <= max_tokens:
                context_parts.append(doc['content'])
                used_tokens += doc['tokens']
            else:
                break
        
        return "\n\n".join(context_parts)
    
    def rank_documents_by_importance(self, docs):
        """基于重要性排序文档"""
        # 简单实现：按相关性分数排序
        return sorted(docs, key=lambda x: x['score'], reverse=True)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def main():
    """主程序"""
    # 初始化组件
    processor = DocumentProcessor()
    retriever = DocumentRetriever(processor.vector_db)
    llm_client = OpenAIClient(api_key="your-api-key")
    
    # 创建 RAG 系统
    rag_system = RAGSystem(retriever, llm_client)
    
    # 测试问答
    query = "公司的加班政策是什么？"
    response, sources = rag_system.generate_response(query)
    
    print("回答：", response)
    print("来源：", sources)

if __name__ == "__main__":
    main()