SaaS 行业观察：上下文工程 - AI 产品的核心竞争力

class RAGSystem:
    def __init__(self):
        self.document_store = VectorDatabase()
        self.retriever = SemanticRetriever()
        self.llm = LargeLanguageModel()
    
    def generate_with_context(self, query):
        # 1. 检索相关文档
        relevant_docs = self.retriever.retrieve(query, top_k=5)
        
        # 2. 构建上下文
        context = self.build_context(relevant_docs, query)
        
        # 3. 生成回答
        response = self.llm.generate(context + query)
        
        return response

def hybrid_retrieve(self, query):
    # 语义检索
    semantic_results = self.semantic_retriever.retrieve(query)
    
    # 关键词检索
    keyword_results = self.keyword_retriever.retrieve(query)
    
    # 融合结果
    combined = self.reciprocal_rank_fusion(semantic_results, keyword_results)
    
    return combined

def rewrite_query(self, query, context):
    # 使用 LLM 重写查询
    rewritten = self.llm.generate(f"""
    原始查询：{query}
    上下文：{context}
    请重写查询以提高检索效果：
    """)
    return rewritten

def multi_hop_retrieve(self, query):
    # 第一轮检索
    first_hop = self.retrieve(query)
    
    # 分析第一轮结果，生成第二轮查询
    second_query = self.generate_follow_up_query(query, first_hop)
    
    # 第二轮检索
    second_hop = self.retrieve(second_query)
    
    # 合并结果
    return first_hop + second_hop

class WorkingMemory:
    def __init__(self, max_tokens=4000):
        self.messages = []
        self.max_tokens = max_tokens
    
    def add_message(self, role, content):
        self.messages.append({"role": role, "content": content})
        self.trim_if_needed()
    
    def trim_if_needed(self):
        while self.count_tokens() > self.max_tokens:
            # 保留系统消息和最近的消息
            self.messages = self.messages[:1] + self.messages[-10:]

class LongTermMemory:
    def __init__(self):
        self.vector_db = VectorDatabase()
        self.graph_db = GraphDatabase()
    
    def store_interaction(self, interaction):
        # 提取关键信息
        entities = self.extract_entities(interaction)
        facts = self.extract_facts(interaction)
        preferences = self.extract_preferences(interaction)
        
        # 存储到向量数据库（用于语义检索）
        self.vector_db.insert(interaction)
        
        # 存储到图数据库（用于关系推理）
        for entity in entities:
            self.graph_db.add_node(entity)
        for fact in facts:
            self.graph_db.add_edge(fact.subject, fact.predicate, fact.object)
    
    def recall(self, query, context):
        # 从向量数据库检索相关记忆
        relevant_memories = self.vector_db.search(query)
        
        # 从图数据库查询相关关系
        related_entities = self.graph_db.query(context.entities)
        
        return relevant_memories + related_entities

class EpisodicMemory:
    def store_episode(self, episode):
        episode_data = {
            "timestamp": datetime.now(),
            "participants": episode.participants,
            "location": episode.location,
            "events": episode.events,
            "emotions": episode.emotions,
            "outcome": episode.outcome
        }
        self.episodes.append(episode_data)
    
    def recall_similar_episodes(self, current_situation):
        # 找到类似的历史情况
        similar = []
        for episode in self.episodes:
            similarity = self.calculate_similarity(current_situation, episode)
            if similarity > 0.7:
                similar.append(episode)
        
        return sorted(similar, key=lambda x: x.similarity, reverse=True)

def summarize_context(self, context, max_length):
    if len(context) <= max_length:
        return context
    
    # 使用 LLM 生成摘要
    summary = self.llm.generate(f"""
    请将以下内容压缩到 {max_length} 字符以内，保留关键信息：
    
    {context}
    """)
    
    return summary

def hierarchical_compression(self, context, max_length):
    # 将上下文分为多个部分
    sections = self.split_into_sections(context)
    
    # 对每个部分生成摘要
    section_summaries = [self.summarize(section) for section in sections]
    
    # 如果还是太长，递归压缩
    combined = "\n".join(section_summaries)
    if len(combined) > max_length:
        return self.hierarchical_compression(combined, max_length)
    
    return combined

def selective_retention(self, context, query):
    # 评估每个部分与查询的相关性
    relevance_scores = []
    for section in context.sections:
        score = self.calculate_relevance(section, query)
        relevance_scores.append((section, score))
    
    # 按相关性排序，保留最相关的部分
    sorted_sections = sorted(relevance_scores, key=lambda x: x[1], reverse=True)
    
    # 在长度限制内选择最相关的部分
    selected = []
    total_length = 0
    for section, score in sorted_sections:
        if total_length + len(section) <= self.max_length:
            selected.append(section)
            total_length += len(section)
    
    return "\n".join(selected)

def inject_context(self, query, user):
    context_parts = []
    
    # 用户档案
    context_parts.append(f"用户信息：{user.profile}")
    
    # 当前项目
    if user.current_project:
        context_parts.append(f"当前项目：{user.current_project.summary}")
    
    # 最近的活动
    recent_activities = self.get_recent_activities(user)
    context_parts.append(f"最近活动：{recent_activities}")
    
    # 相关的知识库内容
    relevant_knowledge = self.retrieve_knowledge(query)
    context_parts.append(f"相关知识：{relevant_knowledge}")
    
    # 组合上下文
    full_context = "\n\n".join(context_parts)
    
    return full_context + "\n\n用户查询：" + query

def conditional_injection(self, query, user):
    context = []
    
    # 分析查询类型
    query_type = self.classify_query(query)
    
    if query_type == "technical":
        # 注入技术文档
        context.append(self.retrieve_technical_docs(query))
    
    elif query_type == "business":
        # 注入业务数据
        context.append(self.retrieve_business_data(query))
    
    elif query_type == "creative":
        # 注入创意灵感
        context.append(self.retrieve_creative_examples(query))
    
    return context