RAG系统设计

1
2
3
4
5
6
7
8
9

class ChapterNode {
    private String chapterId;       
    private String parentChapterId; 
    private String title;           
    private Long startPos;          
    private Long endPos;            
    private List<String> paragraphIds; 
}

1
2
3
4
5
6
7
8
9

class ParagraphNode {
    private String paragraphId;    
    private String chapterId;      
    private String text;           
    private List<String> sentenceIds; 
    private String prevParaId;     
    private String nextParaId;     
}

1
2
3
4
5
6
7
8
9
10

class SentenceNode {
    private String sentenceId;     
    private String paragraphId;    
    private String chapterId;      
    private String text;           
    private float[] vector;        
    private String prevSentId;     
    private String nextSentId;     
}

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

def parse_docx(self, docx_path):
       doc = DocxDocument(docx_path)

       content = []

       image_map = self._extract_images_from_docx(doc)

       def parse_paragraph(paragraph):
           def append_image_link(image_id, has_drawing, target_buffer):
               """Helper to append image link from image_map based on relationship type."""
               rel = doc.part.rels[image_id]
               if rel.is_external:
                   if image_id in image_map and not has_drawing:
                       target_buffer.append(image_map[image_id])
               else:
                   image_part = rel.target_part
                   if image_part in image_map and not has_drawing:
                       target_buffer.append(image_map[image_part])

           def process_run(run, target_buffer):
               
               if hasattr(run.element, "tag") and isinstance(run.element.tag, str) and run.element.tag.endswith("r"):
                   
                   drawing_elements = run.element.findall(
                       ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}drawing"
                   )
                   has_drawing = False
                   for drawing in drawing_elements:
                       blip_elements = drawing.findall(
                           ".//{http://schemas.openxmlformats.org/drawingml/2006/main}blip"
                       )
                       for blip in blip_elements:
                           embed_id = blip.get(
                               "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed"
                           )
                           if embed_id:
                               rel = doc.part.rels.get(embed_id)
                               if rel is not None and rel.is_external:
                                   
                                   if embed_id in image_map:
                                       has_drawing = True
                                       target_buffer.append(image_map[embed_id])
                               else:
                                   
                                   image_part = doc.part.related_parts.get(embed_id)
                                   if image_part in image_map:
                                       has_drawing = True
                                       target_buffer.append(image_map[image_part])
                   
                   shape_elements = run.element.findall(
                       ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}pict"
                   )
                   for shape in shape_elements:
                       
                       shape_image = shape.find(
                           ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}binData"
                       )
                       if shape_image is not None and shape_image.text:
                           image_id = shape_image.get(
                               "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
                           )
                           if image_id and image_id in doc.part.rels:
                               append_image_link(image_id, has_drawing, target_buffer)
                       
                       image_data = shape.find(".//{urn:schemas-microsoft-com:vml}imagedata")
                       if image_data is not None:
                           image_id = image_data.get("id") or image_data.get(
                               "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
                           )
                           if image_id and image_id in doc.part.rels:
                               append_image_link(image_id, has_drawing, target_buffer)
               if run.text.strip():
                   target_buffer.append(run.text.strip())

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

from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding


embed_model = OpenAIEmbedding()




splitter = SemanticSplitterNodeParser(
    buffer_size=1, 
    breakpoint_percentile_threshold=95, 
    embed_model=embed_model
)


nodes = splitter.get_nodes_from_documents(documents)

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

weather_tool = {
  "name": "get_weather", 
  "description": "获取指定城市的天气信息", 
  "parameters": { 
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称，如'北京'、'上海'"
      },
      "date": {
        "type": "string",
        "description": "日期，格式'YYYY-MM-DD'，或'今天'、'明天'",
        "enum": ["今天", "明天", "后天"]
      }
    },
    "required": ["city"] 
  }
}

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 validate_parameters(func_name, params):
    """校验函数参数"""
    
    func_rules = {
        "get_weather": {
            "city": {"type": str, "required": True, "pattern": r"^[\u4e00-\u9fa5]{2,4}$"},
            "date": {"type": str, "required": False, "pattern": r"^\d{4}-\d{2}-\d{2}$"}
        }
    }
    
    if func_name not in func_rules:
        return False, "工具不存在"
    
    rules = func_rules[func_name]
    for param, rule in rules.items():
        
        if rule["required"] and param not in params:
            return False, f"缺失必填参数：{param}"
        
        if param in params and not isinstance(params[param], rule["type"]):
            return False, f"{param}类型错误，需为{rule['type'].__name__}"
        
        if param in params and rule.get("pattern") and not re.match(rule["pattern"], params[param]):
            return False, f"{param}格式错误，需符合规则：{rule['pattern']}"
    
    return True, "参数校验通过"

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

from openai import OpenAI
client = OpenAI()


tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市指定日期的气温",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "中国城市中文全称，如北京、上海"
                    },
                    "date": {
                        "type": "string",
                        "description": "日期，格式为YYYY-MM-DD，可选，默认当天"
                    }
                },
                "required": ["city"]  
            }
        }
    }
]


response = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "查下北京明天的气温"}],
    tools=tools,
    tool_choice="auto",  
    response_format={"type": "json_object"}  
)


tool_call = response.choices[0].message.tool_calls[0]
func_name = tool_call.function.name
func_params = json.loads(tool_call.function.arguments)
print(f"调用工具：{func_name}，参数：{func_params}")

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

class ConversationManager:
    """对话管理器"""

    def __init__(self, max_context_length=4000):
        self.max_context_length = max_context_length  
        self.history = []  
        self.active_context = []  

    def add_exchange(self, user_input, assistant_response):
        """添加一轮对话"""
        self.history.append({
            "user": user_input,
            "assistant": assistant_response,
            "timestamp": time.time()
        })

    def build_context(self, current_query, include_history=True):
        """构建当前查询的上下文"""

        if not include_history or not self.history:
            
            return [{"role": "user", "content": current_query}]

        
        context = []
        context_length = self.estimate_tokens(current_query)

        
        context.insert(0, {"role": "user", "content": current_query})

        
        for exchange in reversed(self.history):
            user_tokens = self.estimate_tokens(exchange["user"])
            assistant_tokens = self.estimate_tokens(exchange["assistant"])

            
            if context_length + user_tokens + assistant_tokens > self.max_context_length:
                break

            
            context.insert(0, {"role": "assistant", "content": exchange["assistant"]})
            context.insert(0, {"role": "user", "content": exchange["user"]})

            context_length += user_tokens + assistant_tokens

        return context

    def estimate_tokens(self, text):
        """粗略估计文本的token数量（实际应用中应使用准确的tokenizer）"""
        
        if self.is_chinese(text):
            return len(text) // 2  
        else:
            words = len(text.split())
            return int(words * 1.3)  

    def is_chinese(self, text):
        """判断文本是否主要为中文"""
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        return chinese_chars / max(len(text), 1) > 0.3

    def clear_history(self):
        """清空对话历史"""
        self.history.clear()
        self.active_context.clear()

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

def time_based_selection(history, current_time, max_items=10):
    """基于时间的选择：越近的对话权重越高"""
    scored_items = []

    for item in history:
        
        time_diff = current_time - item["timestamp"]
        time_score = max(0, 1 - time_diff / 3600)  

        
        total_score = time_score

        scored_items.append((total_score, item))

    
    scored_items.sort(key=lambda x: x[0], reverse=True)
    return [item for score, item in scored_items[:max_items]]

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

def relevance_based_selection(history, current_query, embedding_model, max_items=5):
    """基于与当前查询相关性的选择"""
    if not history:
        return []

    
    query_embedding = embedding_model.encode(current_query)

    scored_items = []

    for item in history:
        
        content = item["user"] + " " + item["assistant"]
        content_embedding = embedding_model.encode(content)

        
        similarity = cosine_similarity([query_embedding], [content_embedding])[0][0]

        scored_items.append((similarity, item))

    
    scored_items.sort(key=lambda x: x[0], reverse=True)
    return [item for similarity, item in scored_items[:max_items]]

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61

class SmartHistorySelector:
    """智能历史选择器"""

    def __init__(self, embedding_model=None):
        self.embedding_model = embedding_model

    def select_history(self, history, current_query, max_context_length=3000):
        """智能选择历史对话"""

        selected = []
        current_length = 0

        
        if history:
            recent = history[-1]
            recent_length = self.estimate_length(recent)
            if recent_length <= max_context_length:
                selected.append(recent)
                current_length += recent_length

        
        if self.embedding_model and len(history) > 1:
            relevant = self.select_by_relevance(history[:-1], current_query, 3)
            for item in relevant:
                item_length = self.estimate_length(item)
                if current_length + item_length <= max_context_length:
                    selected.append(item)
                    current_length += item_length

        
        remaining_space = max_context_length - current_length
        if remaining_space > 100:  
            for item in history:
                if item not in selected:
                    item_length = self.estimate_length(item)
                    if item_length <= remaining_space:
                        selected.append(item)
                        remaining_space -= item_length

        
        selected.sort(key=lambda x: x["timestamp"])
        return selected

    def select_by_relevance(self, history, query, max_items):
        """基于相关性选择"""
        
        query_lower = query.lower()
        scored = []

        for item in history:
            content = (item["user"] + " " + item["assistant"]).lower()
            
            score = sum(1 for word in query_lower.split() if word in content)
            scored.append((score, item))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [item for score, item in scored[:max_items]]

    def estimate_length(self, exchange):
        """估计对话长度"""
        return len(exchange["user"]) + len(exchange["assistant"])

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

class ImportanceBasedCompressor:
    """基于重要性的记忆压缩器"""

    def __init__(self, llm_client):
        self.llm = llm_client

    def compress_memories(self, memories, target_ratio=0.5):
        """压缩记忆，保留重要内容"""

        if len(memories) <= 1:
            return memories  

        
        scored_memories = []
        for memory in memories:
            importance = self.evaluate_importance(memory)
            scored_memories.append((importance, memory))

        
        scored_memories.sort(key=lambda x: x[0], reverse=True)

        
        keep_count = max(1, int(len(memories) * target_ratio))
        compressed_memories = [memory for _, memory in scored_memories[:keep_count]]

        return compressed_memories

    def evaluate_importance(self, memory):
        """评估记忆的重要性"""
        content = memory["content"]
        metadata = memory.get("metadata", {})

        
        importance_score = 0.0

        
        memory_type = metadata.get("type", "")
        if memory_type == "personal_info":
            importance_score += 0.8
        elif memory_type == "preference":
            importance_score += 0.7
        elif memory_type == "fact":
            importance_score += 0.5
        elif memory_type == "conversation":
            importance_score += 0.3

        
        content_length = len(content)
        if 50 <= content_length <= 500:
            importance_score += 0.2
        elif content_length > 500:
            importance_score += 0.1

        
        timestamp = metadata.get("timestamp", 0)
        if timestamp > 0:
            age_days = (time.time() - timestamp) / (24 * 3600)
            recency_score = max(0, 1 - age_days / 30)  
            importance_score += recency_score * 0.5

        
        explicit_importance = metadata.get("importance", 0.5)
        importance_score += explicit_importance * 0.5

        return min(importance_score, 1.0)  

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

from sklearn.cluster import KMeans
import numpy as np

class ClusterBasedCompressor:
    """基于聚类的记忆压缩器"""

    def __init__(self, embedding_model):
        self.embedding_model = embedding_model

    def compress_by_clustering(self, memories, n_clusters=None):
        """通过聚类压缩记忆"""

        if len(memories) <= 3:
            return memories  

        
        if n_clusters is None:
            n_clusters = min(3, len(memories) // 2)

        
        embeddings = []
        for memory in memories:
            embedding = self.embedding_model.encode(memory["content"])
            embeddings.append(embedding)

        embeddings = np.array(embeddings)

        
        kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
        labels = kmeans.fit_predict(embeddings)

        
        compressed_memories = []
        for cluster_id in range(n_clusters):
            
            cluster_memories = [
                memories[i] for i in range(len(memories)) if labels[i] == cluster_id
            ]

            if cluster_memories:
                
                summary = self.summarize_cluster(cluster_memories)
                compressed_memories.append(summary)

        return compressed_memories

    def summarize_cluster(self, cluster_memories):
        """总结一个聚类的记忆"""

        if len(cluster_memories) == 1:
            
            return cluster_memories[0]

        
        all_content = "\n".join([m["content"] for m in cluster_memories])

        
        
        summary_content = f"相关主题的{len(cluster_memories)}条记忆摘要：{all_content[:500]}..."

        
        merged_metadata = {
            "type": "cluster_summary",
            "original_count": len(cluster_memories),
            "compressed": True,
            "timestamp": time.time()
        }

        return {
            "content": summary_content,
            "metadata": merged_metadata
        }

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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

class IncrementalSummarizer:
    """增量总结器"""

    def __init__(self, llm_client, summary_interval=5):
        self.llm = llm_client
        self.summary_interval = summary_interval  
        self.conversation_buffer = []
        self.summaries = []

    def add_conversation(self, user_input, assistant_response):
        """添加对话到缓冲区"""

        self.conversation_buffer.append({
            "user": user_input,
            "assistant": assistant_response,
            "timestamp": time.time()
        })

        
        if len(self.conversation_buffer) >= self.summary_interval:
            self.create_summary()

    def create_summary(self):
        """创建总结"""

        if not self.conversation_buffer:
            return

        
        conversation_text = ""
        for exchange in self.conversation_buffer:
            conversation_text += f"用户: {exchange['user']}\n"
            conversation_text += f"助理: {exchange['assistant']}\n\n"

        
        prompt = f"""
请将以下对话内容总结为一个简短的段落，保留核心信息和重要细节：

对话内容：
{conversation_text}

总结要求：
1. 不超过200字
2. 保留用户的主要需求和助理的关键回答
3. 忽略问候语、重复内容和无关细节

总结：
"""

        summary = self.llm.generate(prompt, max_tokens=200)

        
        self.summaries.append({
            "content": summary,
            "timestamp": time.time(),
            "original_count": len(self.conversation_buffer)
        })

        
        self.conversation_buffer.clear()

    def get_context(self, include_recent=True, include_summaries=True):
        """获取上下文"""

        context_parts = []

        if include_summaries and self.summaries:
            
            for i, summary in enumerate(self.summaries[-3:]):  
                context_parts.append(f"[对话总结 {i+1}]\n{summary['content']}")

        if include_recent and self.conversation_buffer:
            
            for exchange in self.conversation_buffer[-3:]:  
                context_parts.append(f"用户: {exchange['user']}")
                context_parts.append(f"助理: {exchange['assistant']}")

        return "\n\n".join(context_parts)