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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
| """
简单RAG实现
演示：文档加载 → 分割 → 向量化 → 查询
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain.schema import Document
from dotenv import load_dotenv
import os
load_dotenv()
def create_sample_documents() -> list[Document]:
"""创建示例文档"""
texts = [
"""
LangChain是什么？
LangChain是一个开源框架，用于开发由大语言模型(LLM)驱动的应用程序。
它提供了标准化的接口和工具，简化了AI应用的开发过程。
LangChain的核心组件包括：模型、提示词、输出解析器、链、记忆、Agent等。
""",
"""
RAG技术介绍
RAG(Retrieval-Augmented Generation)是检索增强生成技术。
它通过在生成回答之前先检索相关文档，让AI能够访问外部知识库。
RAG的优势包括：准确度高、可更新知识、减少幻觉、成本较低。
RAG的流程：文档加载 → 分割 → 向量化 → 检索 → 生成回答。
""",
"""
向量数据库
向量数据库是专门用于存储和检索向量的数据库。
常见的向量数据库有：Chroma、FAISS、Pinecone、Weaviate等。
向量数据库的核心功能是相似度搜索，通过计算向量间的距离找到最相似的内容。
在RAG应用中，向量数据库用于存储文档的向量表示并快速检索相关内容。
""",
"""
Agent和工具
Agent是能够自主决策和使用工具的AI系统。
与预定义流程的Chain不同，Agent可以根据情况决定使用哪些工具。
常见工具包括：搜索、计算器、数据库查询、API调用等。
Agent的工作流程：接收任务 → 思考 → 决策 → 执行工具 → 观察结果 → 判断是否完成。
"""
]
return [Document(page_content=text.strip()) for text in texts]
def example1_document_processing() -> None:
"""示例1：文档处理流程"""
print("=== 示例1：文档处理 ===\n")
documents = create_sample_documents()
print(f"📄 原始文档数量：{len(documents)}\n")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=50,
separators=["\n\n", "\n", "。", "，", " ", ""]
)
splits = text_splitter.split_documents(documents)
print(f"✂️ 分割后chunks数量：{len(splits)}\n")
print("📋 前两个chunks：")
for i, chunk in enumerate(splits[:2], 1):
print(f"\nChunk {i}:")
print(chunk.page_content[:100] + "...")
def example2_vector_store() -> None:
"""示例2：创建向量数据库"""
print("\n=== 示例2：向量数据库 ===\n")
documents = create_sample_documents()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=50
)
splits = text_splitter.split_documents(documents)
embeddings = OpenAIEmbeddings(
base_url=os.getenv("ALIBABA_BASE_URL"),
api_key=os.getenv("ALIBABA_API_KEY"),
)
print("🔄 正在创建向量数据库...")
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./demo_chroma_db"
)
print("✅ 向量数据库创建完成\n")
print("🔍 测试相似度搜索：")
query = "什么是RAG？"
results = vectorstore.similarity_search(query, k=2)
print(f"\n查询：{query}")
print(f"找到 {len(results)} 个相关文档：\n")
for i, doc in enumerate(results, 1):
print(f"结果 {i}:")
print(doc.page_content[:150])
print()
def example3_simple_qa() -> None:
"""示例3：简单问答系统"""
print("\n=== 示例3：RAG问答系统 ===\n")
documents = create_sample_documents()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=50
)
splits = text_splitter.split_documents(documents)
embeddings = OpenAIEmbeddings(
base_url=os.getenv("ALIBABA_BASE_URL"),
api_key=os.getenv("ALIBABA_API_KEY"),
)
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings
)
llm = ChatOpenAI(
base_url=os.getenv("ALIBABA_BASE_URL"),
api_key=os.getenv("ALIBABA_API_KEY"),
model="qwen-plus",
temperature=0,
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
return_source_documents=True,
verbose=True
)
questions = [
"LangChain有哪些核心组件？",
"RAG技术的优势是什么？",
"Agent和Chain有什么区别？"
]
for question in questions:
print(f"\n{'='*60}")
print(f"❓ 问题：{question}")
print('='*60 + "\n")
result = qa_chain.invoke({"query": question})
print(f"✅ 回答：{result['result']}\n")
print("📚 参考文档：")
for i, doc in enumerate(result['source_documents'], 1):
print(f"\n文档 {i}:")
print(doc.page_content[:100] + "...")
def example4_custom_prompt() -> None:
"""示例4：自定义提示词"""
print("\n\n=== 示例4：自定义RAG提示词 ===\n")
from langchain.prompts import PromptTemplate
documents = create_sample_documents()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=50
)
splits = text_splitter.split_documents(documents)
embeddings = OpenAIEmbeddings(
base_url=os.getenv("ALIBABA_BASE_URL"),
api_key=os.getenv("ALIBABA_API_KEY"),
)
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings
)
llm = ChatOpenAI(
base_url=os.getenv("ALIBABA_BASE_URL"),
api_key=os.getenv("ALIBABA_API_KEY"),
model="qwen-plus",
temperature=0,
)
custom_prompt = PromptTemplate(
template="""
你是一个专业的AI助手。请基于以下参考资料回答问题。
参考资料：
{context}
问题：{question}
回答要求：
1. 只基于参考资料回答，不要编造信息
2. 如果参考资料中没有相关信息，请说"根据提供的资料，我无法回答这个问题"
3. 回答要简洁明了
回答：
""",
input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(),
chain_type_kwargs={"prompt": custom_prompt}
)
question = "向量数据库有哪些？"
print(f"❓ 问题：{question}\n")
result = qa_chain.invoke({"query": question})
print(f"✅ 回答：{result['result']}")
def main() -> None:
"""主函数"""
example1_document_processing()
print("\n" + "="*70 + "\n")
example2_vector_store()
print("\n" + "="*70 + "\n")
example3_simple_qa()
print("\n" + "="*70 + "\n")
example4_custom_prompt()
if __name__ == "__main__":
main()
|