本文最后更新于 2025年6月13日 下午
大模型通信过程
这是一个简单的提问-响应过程。为了解决模型的“幻觉问题”，可以通过多种方式向模型添加有用的信息来增加模型响应结果的可信度
1 . 多轮对话，通过多次提问过程，更加精确的描述用户的实际需求。如Agent
2 . 客户端代理对用户请求进行扩展，最终请求 携带 代理描述信息+原用户请求，向大模型发起请求。如RAG，MCP
3 . 模型端 功能扩展，模型端使用本地工具（一般是一些方法调用）来自行获取扩展信息，增强相应质量。如Function Calling
综上，在从流程上，大模型的扩展会有本地端的功能增强，服务端功能增强两种方式。从内容上，则分为 调用 / 内容 两种。
RAG
RAG 即 检索（Retrieval），增强（Augmented）， 生成 （Generation） 这一过程。其核心原理如下：
- 首先将本地文档进行切分（chunking），并对每个片段进行向量化处理后存入向量数据库；
- 当用户提出问题时，系统会将用户请求向量化，与向量库中的内容进行相似度比对；
- 检索出与问题最相关的若干文档片段，将它们与原始问题组合成 prompt；
- 最终将该增强后的 prompt 提交给大语言模型，由模型基于相关内容生成回答。
简要过程
将目标文档进行分割（chunk过程）
这里对markdown 文件进行简单的处理分割。
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
| def read_data() -> str :
"""
加载目标文件到内存
"""
with open("data.md","r",encoding="utf-8") as f :
return f.read( )
def get_chunks() -> list :
"""
以"\n\n"为分割目标，对文件中的内容进行切割存储
"""
content = read_data()
chunks = content.split("\n\n")
result = []
header = " "
for c in chunks :
if c=="\n\n" :
pass
elif c.startswith("#") :
header += f"{c}\n"
else :
result.append(f"{header}{c}")
header = ""
return result
if __name__ == "__main__" :
chunks = get_chunks()
for c in chunks :
print(c)
|
向量化文档
这里使用了通义千问的模型接口，来进行文档的向量化处理。
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
| import chunks
import requests
import json
from typing import List, Optional
import chromadb
chromadb_client = chromadb.PersistentClient("./chroma.db")
chromadb_collection = chromadb_client.get_or_create_collection(name="my_collection")
class QwenEmbedding:
def __init__(self, api_key: str, base_url: str = "https://dashscope.aliyuncs.com/api/v1"):
"""
初始化通义千问 Embedding 客户端
Args:
api_key: 阿里云 DashScope API Key
base_url: API 基础URL
"""
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_embeddings(self,
texts: List[str],
model: str = "text-embedding-v3",
text_type: str = "document") -> dict:
"""
获取文本嵌入向量
Args:
texts: 要嵌入的文本列表
model: 模型名称
text_type: 文本类型
Returns:
包含嵌入向量的响应字典
"""
url = f"{self.base_url}/services/embeddings/text-embedding/text-embedding"
payload = {
"model": model,
"input": {
"texts": texts
},
"parameters": {
"text_type": text_type
}
}
try:
response = requests.post(url, headers=self.headers, json=payload)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"API 调用失败: {e}")
return None
def get_single_embedding(self,
text: str,
model: str = "text-embedding-v3",
text_type: str = "document") -> Optional[List[float]]:
"""
获取单个文本的嵌入向量
Args:
text: 要嵌入的文本
model: 模型名称
text_type: 文本类型
Returns:
嵌入向量列表，失败时返回 None
"""
result = self.get_embeddings([text], model, text_type)
if result and result.get("output") and result["output"].get("embeddings"):
return result["output"]["embeddings"][0]["embedding"]
return None
def create_db(self) -> None :
"""
将向量化处理后的内容装载到数据库中
"""
for idx, c in enumerate(chunks.get_chunks()) :
embedding = self.get_single_embedding(c)
if embedding :
chromadb_collection.upsert(
ids=[str(idx)],
metadatas=[{"text": c}],
documents=[c],
embeddings=[embedding]
)
def query_db(self , query ):
"""
"""
query_embedding = self.get_single_embedding(query)
results = chromadb_collection.query(query_embeddings=[query_embedding],
n_results=5)
assert results["documents"]
return results
def main():
API_KEY = API_KEY
client = QwenEmbedding(api_key=API_KEY)
client.create_db()
client.query_db(query="克莱尔见到了哪些景象？")
if __name__ == "__main__":
main()
|
创建简易客户端。维护会话过程
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
| import requests
import json
import embedding
class Client (object):
def __init__(self, api_key: str, base_url: str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"):
"""
初始化请求信息
Args :
api_key : 千问模型api
base_url : 访问的基础接口
"""
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_message(self, input_text: str) -> dict:
"""
接收用户请求，进行向量查询，获取近似结果
Args :
input_text : 用户输入的请求
"""
db = embedding.QwenEmbedding(self.api_key)
db.create_db()
return db.query_db(query=input_text)
def create_message_body(self, input_text: str) -> dict:
"""
构造发送给大模型的消息体
Args :
input_text : 用户输入请求
"""
return {
"model": "qwen3-8b",
"input": {
"messages": [
{"role": "system", "content": "你一个 helpful 的助手"},
{"role": "user", "content": input_text}
]
},
"parameters": {
"temperature": 0.8,
"enable_thinking": False
}
}
def run(self) :
"""
执行函数，通过while 持续监听用户对话。
"""
print("欢迎，请输入问题：")
while True:
input_text = input()
if input_text in ["","quit","exit","q"]:
print("Bye")
break
rag_message = self.create_message(input_text)
for i in rag_message["metadatas"][0]:
input_text += i["text"]+"\n"
message_body=self.create_message_body(input_text)
resp = requests.post(self.base_url, headers=self.headers, json=message_body)
print(resp.json()["output"]["choices"][0]["message"]["content"])
if __name__ == "__main__":
client = Client("API-KEY")
client.run()
|
执行
执行程序后，程序会根据用户提问检索本地知识库，并将相关的内容补充到向大模型发送的请求中。大模型依靠这些相关知识可以进行更全面的知识解答。
由此，我们可以知道，RAG 提供了从客户端到大模型接口间的能力扩展。通过本地向量数据库的数据对比来完成相关信息的检索，帮助大模型完善相关知识来得到更全面的解答