Agent 上下文工程实战：从 128K 窗口到 2M 窗口的工程权衡与最佳实践

from langchain.chat_models import ChatOpenAI
from langchain.chains import SummarizationChain

class ConversationCompressor:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-5.5-mini", temperature=0)
        self.summary_chain = SummarizationChain.from_llm(self.llm)
        self.window_size = 10  # 每 10 轮为一个窗口
    
    def compress(self, conversation_history):
        """将对话历史压缩为摘要"""
        summaries = []
        
        # 按窗口分割对话历史
        for i in range(0, len(conversation_history), self.window_size):
            window = conversation_history[i:i + self.window_size]
            
            # 生成窗口摘要
            summary = self.summary_chain.run(
                input=f"Summarize the following conversation in 3-5 sentences:\\n\\n{window}"
            )
            summaries.append(summary)
        
        # 拼接所有摘要
        compressed_context = "\\n".join(summaries)
        return compressed_context

import json
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

class KeyInfoExtractor:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-5.5-mini", temperature=0)
        self.prompt = ChatPromptTemplate.from_template("""
        Extract key information from the following conversation:
        - User preferences
        - Decisions made
        - Action items
        - Important facts
        
        Output as JSON:
        {{
            "preferences": [...],
            "decisions": [...],
            "action_items": [...],
            "facts": [...]
        }}
        
        Conversation:
        {conversation}
        """)
    
    def extract(self, conversation_history):
        """提取关键信息"""
        chain = self.prompt | self.llm
        response = chain.invoke({"conversation": conversation_history})
        
        # 解析 JSON
        key_info = json.loads(response.content)
        return key_info
    
    def format_for_context(self, key_info):
        """将关键信息格式化为上下文"""
        formatted = []
        
        if key_info.get("preferences"):
            formatted.append("User Preferences:")
            for pref in key_info["preferences"]:
                formatted.append(f"- {pref}")
        
        if key_info.get("decisions"):
            formatted.append("\\nDecisions Made:")
            for decision in key_info["decisions"]:
                formatted.append(f"- {decision}")
        
        if key_info.get("action_items"):
            formatted.append("\\nAction Items:")
            for item in key_info["action_items"]:
                formatted.append(f"- {item}")
        
        return "\\n".join(formatted)

class HierarchicalCompressor:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-5.5-mini", temperature=0)
        self.summary_chain = SummarizationChain.from_llm(self.llm)
    
    def compress(self, conversation_history):
        """分层压缩对话历史"""
        # 第一层：段落级摘要（每 5 轮）
        paragraph_summaries = []
        for i in range(0, len(conversation_history), 5):
            paragraph = conversation_history[i:i + 5]
            summary = self.summary_chain.run(
                input=f"Summarize in 2-3 sentences:\\n\\n{paragraph}"
            )
            paragraph_summaries.append(summary)
        
        # 第二层：章节级摘要（每 3 个段落摘要）
        section_summaries = []
        for i in range(0, len(paragraph_summaries), 3):
            section = "\\n".join(paragraph_summaries[i:i + 3])
            summary = self.summary_chain.run(
                input=f"Summarize in 3-4 sentences:\\n\\n{section}"
            )
            section_summaries.append(summary)
        
        # 第三层：全局摘要
        full_summary = "\\n".join(section_summaries)
        global_summary = self.summary_chain.run(
            input=f"Summarize in 5-7 sentences:\\n\\n{full_summary}"
        )
        
        return global_summary

class DynamicContextLoader:
    def __init__(self):
        self.loaded_context = set()
    
    def load_context(self, user_query, conversation_state):
        """根据用户查询和对话状态动态加载上下文"""
        context = []
        
        # 第一轮：加载用户基本信息
        if "user_info" not in self.loaded_context:
            user_info = self.get_user_info(conversation_state.user_id)
            context.append(f"## 用户信息\\n{user_info}")
            self.loaded_context.add("user_info")
        
        # 根据查询内容加载相关信息
        if "退款" in user_query and "refund_policy" not in self.loaded_context:
            refund_policy = self.get_refund_policy()
            context.append(f"## 退款政策\\n{refund_policy}")
            self.loaded_context.add("refund_policy")
        
        if "订单" in user_query and "order_history" not in self.loaded_context:
            order_history = self.get_order_history(conversation_state.user_id)
            context.append(f"## 订单历史\\n{order_history}")
            self.loaded_context.add("order_history")
        
        return "\\n".join(context)

class ContextVersionControl:
    def __init__(self):
        self.context_versions = {}
        self.current_version = "v1.0"
    
    def update_context(self, new_context, reason):
        """更新上下文并记录版本"""
        version = f"v{len(self.context_versions) + 1}.0"
        self.context_versions[version] = {
            "context": new_context,
            "reason": reason,
            "date": datetime.now().isoformat()
        }
        self.current_version = version
        return version
    
    def rollback(self, version):
        """回滚到指定版本"""
        if version in self.context_versions:
            self.current_version = version
            return self.context_versions[version]["context"]
        raise ValueError(f"Version {version} not found")
    
    def compare_versions(self, v1, v2):
        """比较两个版本的差异"""
        context1 = self.context_versions[v1]["context"]
        context2 = self.context_versions[v2]["context"]
        return diff(context1, context2)

class ContextMonitor:
    def __init__(self):
        self.metrics = {
            "retrieval_accuracy": [],
            "compression_ratio": [],
            "task_completion_rate": []
        }
    
    def log_retrieval_accuracy(self, query, retrieved_chunks, relevant_chunks):
        """记录检索准确率"""
        accuracy = len(set(retrieved_chunks) & set(relevant_chunks)) / len(retrieved_chunks)
        self.metrics["retrieval_accuracy"].append({
            "query": query,
            "accuracy": accuracy,
            "timestamp": datetime.now().isoformat()
        })
        return accuracy
    
    def log_compression_ratio(self, original_length, compressed_length):
        """记录压缩比"""
        ratio = original_length / compressed_length
        self.metrics["compression_ratio"].append({
            "ratio": ratio,
            "original_length": original_length,
            "compressed_length": compressed_length,
            "timestamp": datetime.now().isoformat()
        })
        return ratio
    
    def log_task_completion(self, task_id, completed, context_version):
        """记录任务完成率"""
        self.metrics["task_completion_rate"].append({
            "task_id": task_id,
            "completed": completed,
            "context_version": context_version,
            "timestamp": datetime.now().isoformat()
        })
    
    def get_report(self):
        """生成上下文性能报告"""
        report = {
            "avg_retrieval_accuracy": sum(m["accuracy"] for m in self.metrics["retrieval_accuracy"]) / len(self.metrics["retrieval_accuracy"]),
            "avg_compression_ratio": sum(m["ratio"] for m in self.metrics["compression_ratio"]) / len(self.metrics["compression_ratio"]),
            "task_completion_rate": sum(1 for m in self.metrics["task_completion_rate"] if m["completed"]) / len(self.metrics["task_completion_rate"])
        }
        return report