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

from openai import OpenAI

client = OpenAI(
    base_url="https://api.openai.com/v1/",  # 更换为不同提供商的URL
    api_key="your-api-key"
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手。"},
        {"role": "user", "content": "请解释什么是RAG"}
    ]
)

print(response.choices[0].message.content)

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

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "请解释RAG"}],
    stream=True
)

for chunk in response:
    # 部分模型有reasoning_content（思考过程）
    if hasattr(chunk.choices[0].delta, 'reasoning_content'):
        reasoning = chunk.choices[0].delta.reasoning_content
        if reasoning:
            print(reasoning, end='')
    # 正式回复内容
    content = chunk.choices[0].delta.content
    if content:
        print(content, end='')

1
2
3
4
5
6

response = client.chat.completions.create(
    model="gpt-4o",
    temperature=0.0,     # 确定性输出
    max_tokens=500,
    messages=[{"role": "user", "content": "列出Python的5个核心特性"}]
)

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

from pydantic import BaseModel

class MovieReview(BaseModel):
    title: str
    rating: float
    summary: str
    keywords: list[str]

completion = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "请分析用户提供的电影评论。"},
        {"role": "user", "content": "《流浪地球2》特效震撼，剧情宏大，是中国科幻电影的里程碑。"}
    ],
    response_format=MovieReview
)

result = completion.choices[0].message.parsed
print(result.title)     # 流浪地球2
print(result.rating)    # 9.0
print(result.keywords)  # ['科幻', '特效', '中国电影']

1
2
3
4
5
6
7
8

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "请以JSON格式返回分析结果，包含title、rating、summary字段。"},
        {"role": "user", "content": "分析《流浪地球2》"}
    ]
)

1
2
3
4
5
6

角色：你是一位{领域}专家。
任务：{具体任务描述}
要求：
1. {约束条件1}
2. {约束条件2}
输出格式：{期望格式}

1
2
3

任务：{任务描述}
约束：{具体约束}
输入：{用户输入}

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

messages = [
    {"role": "system", "content": "你是一个情感分析助手。"},
    # 示例1
    {"role": "user", "content": "这家餐厅的菜太难吃了"},
    {"role": "assistant", "content": "负面"},
    # 示例2
    {"role": "user", "content": "服务态度很好，环境也不错"},
    {"role": "assistant", "content": "正面"},
    # 示例3
    {"role": "user", "content": "价格一般，味道还行"},
    {"role": "assistant", "content": "中性"},
    # 实际问题
    {"role": "user", "content": "性价比超高，下次还来！"}
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    temperature=0
)
# 输出：正面

1
2
3
4
5
6
7
8

messages = [
    {"role": "user", "content": """
一个商店有100个苹果。第一天卖出了30%，第二天卖出了剩余的50%。
还剩多少个苹果？

请一步一步地思考，然后给出最终答案。
"""}
]

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

import collections

def self_consistency(question, n_samples=5):
    answers = []
    for _ in range(n_samples):
        response = client.chat.completions.create(
            model="gpt-4o",
            temperature=0.7,  # 需要一定随机性以产生多样答案
            messages=[
                {"role": "user", "content": f"{question}\n请一步步思考并给出最终答案。"}
            ]
        )
        # 提取最终答案（需要根据具体格式解析）
        answers.append(extract_final_answer(response))

    # 投票：选出出现次数最多的答案
    counter = collections.Counter(answers)
    return counter.most_common(1)[0][0]

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

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的天气信息",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京"
                    }
                },
                "required": ["city"],
                "additionalProperties": False
            }
        }
    }
]

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

messages = [{"role": "user", "content": "北京今天天气怎么样？"}]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools
)

# 检查模型是否决定调用函数
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    print(tool_call.function.name)       # "get_weather"
    print(tool_call.function.arguments)  # '{"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
27

import json

def get_weather(city):
    # 实际中调用天气API
    return {"city": city, "temperature": "25°C", "condition": "晴"}

# 解析参数并执行
args = json.loads(tool_call.function.arguments)
result = get_weather(**args)

# 将函数结果返回给模型
messages.append(response.choices[0].message)  # 加入模型的tool_call消息
messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": json.dumps(result, ensure_ascii=False)
})

# 模型基于函数结果生成最终回复
final_response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools
)

print(final_response.choices[0].message.content)
# 输出：北京今天天气晴朗，温度25°C，适合外出活动。

1

用户提问 → 向量化查询 → 在向量数据库中检索相似文档 → 将文档+问题送入LLM → 生成回答

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import numpy as np

def cosine_similarity(vec1, vec2):
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

# 创建嵌入
response = client.embeddings.create(
    model="text-embedding-ada-002",
    input="什么是人工智能？"
)
query_embedding = response.data[0].embedding

1
2
3
4
5
6
7
8
9

import fitz  # PyMuPDF

def extract_text_from_pdf(pdf_path):
    mypdf = fitz.open(pdf_path)
    all_text = ""
    for page_num in range(mypdf.page_count):
        page = mypdf[page_num]
        all_text += page.get_text("text")
    return all_text

1
2
3
4
5

def chunk_text(text, n=1000, overlap=200):
    chunks = []
    for i in range(0, len(text), n - overlap):
        chunks.append(text[i:i + n])
    return chunks

1
2
3

def create_embeddings(text, model="text-embedding-ada-002"):
    response = client.embeddings.create(model=model, input=text)
    return response

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

def semantic_search(query, text_chunks, embeddings, k=5):
    query_embedding = create_embeddings(query).data[0].embedding
    similarity_scores = []
    for i, chunk_embedding in enumerate(embeddings):
        score = cosine_similarity(
            np.array(query_embedding),
            np.array(chunk_embedding.embedding)
        )
        similarity_scores.append((i, score))
    similarity_scores.sort(key=lambda x: x[1], reverse=True)
    top_indices = [idx for idx, _ in similarity_scores[:k]]
    return [text_chunks[idx] for idx in top_indices]

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

def generate_response(query, context, model="gpt-4o"):
    system_prompt = "你是一名AI助手，只根据提供的上下文回答问题。"
    user_prompt = f"Context:\n{context}\n\nQuestion: {query}\n\n请基于上下文回答。"

    response = client.chat.completions.create(
        model=model,
        temperature=0,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    )
    return response.choices[0].message.content

1
2
3
4
5
6
7

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("什么是人工智能？")
print(response)

1
2
3
4
5
6
7
8

from llama_index.core import Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o", temperature=0)
Settings.embed_model = OpenAIEmbedding(model_name="text-embedding-ada-002")
Settings.text_splitter = SentenceSplitter(chunk_size=1000, chunk_overlap=200)

1
2
3
4
5
6
7
8
9

import faiss
from llama_index.vector_stores.faiss import FaissVectorStore
from llama_index.core import StorageContext

d = 1536  # embedding维度
faiss_index = faiss.IndexFlatL2(d)
vector_store = FaissVectorStore(faiss_index=faiss_index)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(documents, storage_context=storage_context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from llama_index.core.memory import ChatMemoryBuffer

memory = ChatMemoryBuffer.from_defaults(token_limit=15000)
chat_engine = index.as_chat_engine(
    chat_mode="context",
    memory=memory,
    system_prompt="你是一个有帮助的AI助手。"
)
response = chat_engine.chat("什么是深度学习？")
# 多轮对话自动保留上下文
response2 = chat_engine.chat("它有哪些应用？")

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

def compute_breakpoints(similarities, method="percentile", threshold=90):
    if method == "percentile":
        # 百分位法：相似度低于第X百分位的位置为断点
        threshold_value = np.percentile(similarities, threshold)
    elif method == "standard_deviation":
        # 标准差法：相似度低于 均值-X*标准差 的位置
        mean = np.mean(similarities)
        std_dev = np.std(similarities)
        threshold_value = mean - (threshold * std_dev)
    elif method == "interquartile":
        # IQR法：使用四分位距检测异常低值
        q1, q3 = np.percentile(similarities, [25, 75])
        threshold_value = q1 - 1.5 * (q3 - q1)

    return [i for i, sim in enumerate(similarities) if sim < threshold_value]

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

def context_enriched_search(query, text_chunks, embeddings, k=1, context_size=1):
    query_embedding = create_embeddings(query).data[0].embedding
    similarity_scores = []

    for i, chunk_embedding in enumerate(embeddings):
        score = cosine_similarity(
            np.array(query_embedding),
            np.array(chunk_embedding.embedding)
        )
        similarity_scores.append((i, score))

    similarity_scores.sort(key=lambda x: x[1], reverse=True)
    top_index = similarity_scores[0][0]

    # 取前后context_size个块
    start = max(0, top_index - context_size)
    end = min(len(text_chunks), top_index + context_size + 1)

    return [text_chunks[i] for i in range(start, end)]

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

def generate_chunk_header(chunk, model="gpt-4o"):
    response = client.chat.completions.create(
        model=model,
        temperature=0,
        messages=[
            {"role": "system", "content": "Generate a concise and informative title for the given text."},
            {"role": "user", "content": chunk}
        ]
    )
    return response.choices[0].message.content.strip()

# 检索时综合标题和内容的相似度
def semantic_search(query, chunks, k=5):
    query_embedding = create_embeddings(query)
    similarities = []
    for chunk in chunks:
        sim_text = cosine_similarity(query_embedding, chunk["embedding"])
        sim_header = cosine_similarity(query_embedding, chunk["header_embedding"])
        avg_similarity = (sim_text + sim_header) / 2
        similarities.append((chunk, avg_similarity))
    similarities.sort(key=lambda x: x[1], reverse=True)
    return [x[0] for x in similarities[:k]]

 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

def generate_questions(text_chunk, num_questions=5, model="gpt-4o"):
    system_prompt = "你是一名擅长从文本中提炼关键信息并生成相关问题的专家。"
    user_prompt = f"请根据以下文本内容，生成 {num_questions} 个不同的问题：\n\n{text_chunk}"

    response = client.chat.completions.create(
        model=model,
        temperature=0.7,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
    )
    # 解析返回的问题列表...
    return questions

# 处理文档时，为每个块生成问题并存入向量库
for i, chunk in enumerate(text_chunks):
    chunk_embedding = create_embeddings(chunk)
    vector_store.add_item(text=chunk, embedding=chunk_embedding,
                          metadata={"type": "chunk", "index": i})

    questions = generate_questions(chunk, num_questions=3)
    for question in questions:
        q_embedding = create_embeddings(question)
        vector_store.add_item(text=question, embedding=q_embedding,
                              metadata={"type": "question", "chunk_index": i,
                                        "original_chunk": chunk})

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

def rewrite_query(original_query, model="gpt-4o"):
    """将查询改写得更具体、更详细"""
    system_prompt = "你是一名擅长优化检索查询的AI助手。"
    user_prompt = f"请将下列查询改写为更具体、更详细的表达：\n原始查询：{original_query}"
    response = client.chat.completions.create(
        model=model, temperature=0.0,
        messages=[{"role": "system", "content": system_prompt},
                  {"role": "user", "content": user_prompt}]
    )
    return response.choices[0].message.content.strip()

def generate_step_back_query(original_query, model="gpt-4o"):
    """生成更宽泛的后退查询"""
    system_prompt = "你的任务是将具体查询改写为更宽泛的问题，以获取背景信息。"
    # ...
    return step_back_query

def decompose_query(original_query, num_subqueries=4, model="gpt-4o"):
    """将复杂查询拆解为子问题"""
    system_prompt = "你的任务是把复杂查询分解为若干简单子问题。"
    # ...
    return sub_queries  # List[str]

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

def rerank_with_llm(query, chunks, model="gpt-4o"):
    scored_chunks = []
    for chunk in chunks:
        prompt = f"""请根据查询对以下文本的相关性打分（0-10分）。
查询：{query}
文本：{chunk['text']}
只返回数字分数。"""
        response = client.chat.completions.create(
            model=model, temperature=0,
            messages=[{"role": "user", "content": prompt}]
        )
        score = float(response.choices[0].message.content.strip())
        scored_chunks.append({**chunk, "rerank_score": score})

    scored_chunks.sort(key=lambda x: x["rerank_score"], reverse=True)
    return scored_chunks

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

import jieba

def rerank_with_keywords(query, chunks):
    query_keywords = set(jieba.cut(query))
    scored_chunks = []
    for chunk in chunks:
        chunk_words = list(jieba.cut(chunk['text']))
        score = 0
        for keyword in query_keywords:
            if keyword in chunk_words:
                # 位置越靠前，分数越高
                pos = chunk_words.index(keyword) / len(chunk_words)
                freq = chunk_words.count(keyword)
                score += (1 - pos) * 0.5 + freq * 0.5
        scored_chunks.append({**chunk, "keyword_score": score})

    scored_chunks.sort(key=lambda x: x["keyword_score"], reverse=True)
    return scored_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

def calculate_chunk_values(similarities, threshold=0.5, irrelevant_penalty=-0.2):
    """将相似度转换为收益值：相关块=正分，不相关块=负分"""
    values = []
    for sim in similarities:
        if sim >= threshold:
            values.append(sim)       # 相关块贡献正分
        else:
            values.append(irrelevant_penalty)  # 不相关块扣分
    return values

def find_best_segments(values, max_segments=3, min_length=1):
    """使用最大子数组变体寻找最优连续段落"""
    segments = []
    n = len(values)

    for _ in range(max_segments):
        best_sum = float('-inf')
        best_start = best_end = 0
        current_sum = 0
        start = 0

        for i in range(n):
            current_sum += values[i]
            if current_sum > best_sum and (i - start + 1) >= min_length:
                best_sum = current_sum
                best_start = start
                best_end = i
            if current_sum < 0:
                current_sum = 0
                start = i + 1

        if best_sum <= 0:
            break

        segments.append((best_start, best_end, best_sum))
        # 将已选区间置零，避免重复选择
        for i in range(best_start, best_end + 1):
            values[i] = 0

    return segments

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

def compress_chunk(chunk, query, compression_type="selective", model="gpt-4o"):
    if compression_type == "selective":
        prompt = f"只保留以下文本中与问题直接相关的句子：\n问题：{query}\n文本：{chunk}"
    elif compression_type == "summary":
        prompt = f"用2-3个简短句子总结以下文本中与问题相关的内容：\n问题：{query}\n文本：{chunk}"
    elif compression_type == "extraction":
        prompt = f"从以下文本中提取与问题相关的关键事实和数据：\n问题：{query}\n文本：{chunk}"

    response = client.chat.completions.create(
        model=model, temperature=0,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content.strip()

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

def store_feedback(feedback_store, query, response, feedback_score, context):
    """存储用户反馈"""
    feedback_store.append({
        "query": query,
        "response": response,
        "feedback": feedback_score,  # 1=满意, 0=不满意
        "context": context
    })

def fine_tune_index(vector_store, feedback_store):
    """用高质量反馈增强向量库"""
    for item in feedback_store:
        if item["feedback"] >= 4:  # 高分反馈
            # 将高质量QA对作为新文档加入向量库
            qa_text = f"Q: {item['query']}\nA: {item['response']}"
            embedding = create_embeddings(qa_text)
            vector_store.add_item(
                text=qa_text,
                embedding=embedding,
                metadata={"type": "feedback", "relevance_score": 1.2}  # 提升权重
            )

 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

def classify_query(query, model="gpt-4o"):
    """将查询分类为四种类型"""
    prompt = f"""将以下查询分类为：factual/analytical/opinion/contextual
查询：{query}
只返回类型名称。"""
    response = client.chat.completions.create(
        model=model, temperature=0,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content.strip().lower()

def adaptive_retrieval(query, vector_store):
    query_type = classify_query(query)

    if query_type == "factual":
        # 精确匹配，少量高相似度结果
        return vector_store.similarity_search(query_embedding, k=3)
    elif query_type == "analytical":
        # 多角度检索，结合子查询分解
        sub_queries = decompose_query(query)
        results = []
        for sq in sub_queries:
            results.extend(vector_store.similarity_search(sq_embedding, k=2))
        return deduplicate(results)
    elif query_type == "opinion":
        # 多样化来源
        return vector_store.similarity_search(query_embedding, k=5)
    elif query_type == "contextual":
        # 上下文增强检索
        return context_enriched_search(query, ...)

 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

def self_rag(query, vector_store):
    # 1. 判断是否需要检索
    needs_retrieval = determine_if_retrieval_needed(query)
    if not needs_retrieval:
        return generate_without_retrieval(query)

    # 2. 检索文档
    results = vector_store.similarity_search(query_embedding, k=5)

    best_response = None
    best_utility = 0

    for result in results:
        # 3. 评估相关性
        relevance = evaluate_relevance(query, result["text"])  # "relevant"/"irrelevant"
        if relevance != "relevant":
            continue

        # 4. 生成回复
        response = generate_response(query, result["text"])

        # 5. 评估支撑度
        support = assess_support(response, result["text"])  # "fully"/"partially"/"no"

        # 6. 评估效用
        utility = rate_utility(query, response)  # 1-5分

        if utility > best_utility:
            best_utility = utility
            best_response = response

    return best_response

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

def generate_propositions(text_chunk, model="gpt-4o"):
    prompt = """将以下文本分解为独立的、自包含的事实命题。
每个命题应该：
1. 表达一个完整的事实
2. 脱离上下文也能理解
3. 尽可能简洁

文本：{text_chunk}

请以编号列表形式输出所有命题。"""
    response = client.chat.completions.create(
        model=model, temperature=0,
        messages=[{"role": "user", "content": prompt}]
    )
    return parse_propositions(response.choices[0].message.content)

def evaluate_proposition(proposition, original_chunk, model="gpt-4o"):
    """评估命题的质量：准确性、清晰性、完整性、简洁性"""
    # 返回0-5的评分
    ...

 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 extract_content_from_pdf(pdf_path):
    """提取文本和图像"""
    doc = fitz.open(pdf_path)
    content = {"text": "", "images": []}

    for page in doc:
        content["text"] += page.get_text("text")
        for img_info in page.get_images():
            xref = img_info[0]
            img_data = doc.extract_image(xref)
            content["images"].append(img_data)

    return content

def generate_image_caption(image_base64, model="gpt-4o"):
    """使用视觉模型为图像生成描述"""
    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "请详细描述这张图片的内容。"},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/png;base64,{image_base64}"
                }}
            ]
        }]
    )
    return response.choices[0].message.content

 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
import jieba

def fusion_retrieval(query, text_chunks, embeddings, alpha=0.5):
    """alpha控制向量搜索与BM25的权重"""
    # 向量搜索分数
    query_embedding = create_embeddings(query)
    vector_scores = [cosine_similarity(query_embedding, emb) for emb in embeddings]

    # BM25分数
    tokenized_chunks = [list(jieba.cut(chunk)) for chunk in text_chunks]
    bm25 = BM25Okapi(tokenized_chunks)
    query_tokens = list(jieba.cut(query))
    bm25_scores = bm25.get_scores(query_tokens)

    # 归一化
    vector_scores = normalize(vector_scores)
    bm25_scores = normalize(bm25_scores)

    # 加权融合
    combined = alpha * vector_scores + (1 - alpha) * bm25_scores

    top_indices = np.argsort(combined)[-k:][::-1]
    return [text_chunks[i] for i in top_indices]

 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

import networkx as nx

def extract_concepts(text_chunk, model="gpt-4o"):
    """使用LLM从文本中提取概念和关系"""
    prompt = f"""从以下文本中提取关键概念及其关系。
格式：概念A -> 关系 -> 概念B
文本：{text_chunk}"""
    response = client.chat.completions.create(model=model, ...)
    return parse_concepts(response.choices[0].message.content)

def build_knowledge_graph(text_chunks):
    """构建知识图谱"""
    graph = nx.Graph()
    for chunk in text_chunks:
        concepts = extract_concepts(chunk)
        for concept_a, relation, concept_b in concepts:
            graph.add_node(concept_a)
            graph.add_node(concept_b)
            graph.add_edge(concept_a, concept_b, relation=relation)
    return graph

def traverse_graph(graph, query_concepts, max_depth=2):
    """从查询相关节点出发，BFS遍历图获取相关信息"""
    relevant_nodes = set()
    for concept in query_concepts:
        if concept in graph:
            # 广度优先搜索，限制深度
            for node in nx.bfs_tree(graph, concept, depth_limit=max_depth):
                relevant_nodes.add(node)
    return relevant_nodes

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

def generate_page_summary(page_text, model="gpt-4o"):
    """为每页生成摘要"""
    response = client.chat.completions.create(
        model=model, temperature=0,
        messages=[{"role": "user", "content": f"请用2-3句话总结：\n{page_text}"}]
    )
    return response.choices[0].message.content

def retrieve_hierarchically(query, summary_store, detailed_store, top_summaries=3, top_chunks=5):
    """分层检索"""
    # 第一级：在摘要中搜索，定位相关页面
    query_embedding = create_embeddings(query)
    summary_results = summary_store.similarity_search(query_embedding, k=top_summaries)

    # 获取相关页面的ID
    relevant_pages = [r["metadata"]["page_id"] for r in summary_results]

    # 第二级：只在相关页面的详细块中搜索
    filtered_results = detailed_store.similarity_search(
        query_embedding, k=top_chunks,
        filter={"page_id": {"$in": relevant_pages}}
    )

    return filtered_results

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

def generate_hypothetical_document(query, model="gpt-4o"):
    """生成假设性回答"""
    prompt = f"请写一段可能回答以下问题的文档内容（不需要完全准确）：\n{query}"
    response = client.chat.completions.create(
        model=model, temperature=0.7,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

def hyde_rag(query, vector_store):
    # 1. 生成假设文档
    hypothetical_doc = generate_hypothetical_document(query)

    # 2. 用假设文档的嵌入检索
    hyp_embedding = create_embeddings(hypothetical_doc)
    results = vector_store.similarity_search(hyp_embedding, k=5)

    # 3. 基于真实检索结果生成回复
    context = "\n".join([r["text"] for r in results])
    return generate_response(query, context)

 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

def evaluate_document_relevance(query, document, model="gpt-4o"):
    """评估文档与查询的相关性（0-1分）"""
    prompt = f"评估以下文档与查询的相关性（0-1分）：\n查询：{query}\n文档：{document}"
    response = client.chat.completions.create(model=model, ...)
    return float(response.choices[0].message.content.strip())

def crag_pipeline(query, vector_store):
    # 1. 检索文档
    results = vector_store.similarity_search(query_embedding, k=5)

    # 2. 评估每篇文档的相关性
    for result in results:
        result["relevance"] = evaluate_document_relevance(query, result["text"])

    # 3. 根据相关性分级处理
    high_relevance = [r for r in results if r["relevance"] >= 0.7]
    medium_relevance = [r for r in results if 0.3 <= r["relevance"] < 0.7]

    if high_relevance:
        # 高相关性：直接使用
        context = "\n".join([r["text"] for r in high_relevance])
    elif medium_relevance:
        # 中等相关性：精炼知识 + 网络搜索补充
        refined = [refine_knowledge(query, r["text"]) for r in medium_relevance]
        web_results = web_search(rewrite_search_query(query))
        context = "\n".join(refined + web_results)
    else:
        # 低相关性：完全依赖网络搜索
        web_results = web_search(query)
        context = "\n".join(web_results)

    return generate_response(query, context)

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

def define_state(query, context, history):
    """状态 = 当前查询 + 已有上下文 + 操作历史"""
    return {"query": query, "context": context, "history": history}

def define_action_space():
    """可选操作"""
    return ["rewrite_query", "expand_context", "filter_context", "generate_response"]

def calculate_reward(response, reference_answer):
    """奖励 = 回复与参考答案的余弦相似度"""
    resp_emb = create_embeddings(response)
    ref_emb = create_embeddings(reference_answer)
    return cosine_similarity(resp_emb, ref_emb)

def policy_network(state, epsilon=0.1):
    """epsilon-greedy策略：以epsilon概率随机探索，否则选最优"""
    if random.random() < epsilon:
        return random.choice(define_action_space())
    else:
        return select_best_action(state)

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

def training_loop(queries, reference_answers, num_episodes=100):
    for episode in range(num_episodes):
        for query, ref_answer in zip(queries, reference_answers):
            state = define_state(query, "", [])
            total_reward = 0

            for step in range(max_steps):
                action = policy_network(state, epsilon)
                new_state, reward = rl_step(state, action)
                total_reward += reward
                # 更新策略网络...
                state = new_state

                if action == "generate_response":
                    break