当前位置: 首页 > news >正文

LangChain RAG 实战

文档准备

# 存放所有文档的路径
path_list = ["../note/太上老君的炉子.txt","../note/中国第一颗原子弹.txt","../note/大明王朝.txt","../note/人工智能发展史.txt"
]

文档加载和分割

# 存放分割后所有的文档块
docs_list = []
# 创建分割器 按字符分割
ext_splitter = CharacterTextSplitter(chunk_size=100, # 每个文档块的最大字符数chunk_overlap=10 # 相邻文档块间的重叠字符数
)
# 遍历文档
for path in path_list:# 加载文档documents = TextLoader(path, encoding="utf-8").load()# 分割docs = ext_splitter.split_documents(documents)# 存储docs_list += docs

初始化 Embedding 模型

embeddings = DashScopeEmbeddings(model="text-embedding-v1",  # 通义千问官方 Embedding 模型dashscope_api_key=key,
)

初始化 向量存储数据库实例

vector_store = Chroma(collection_name="my_docs",  # 集合名称embedding_function=embeddings,  # 嵌入模型persist_directory="../chroma_db"  # 持久化目录
)

文档列表存入向量数据库

uuids = [str(uuid4()) for _ in range(len(docs_list))]
vector_store.add_documents(documents=docs_list, ids=uuids)

创建检索器

retriever = vector_store.as_retriever()

处理检索结果

def format_docs(docs):# 将文档中的page_content属性以换行符连接起来return "\\n\\n".join(doc.page_content for doc in docs)

创建一个数据处理管道

用于获取context内容

  • 输入:{"question": "问题内容", ...}
  • 输出:文档1:xxxx...\\n\\n文档2:xxxx...
get_context = itemgetter("question") | retriever | format_docs

数据流整合

将context拼接到question之后

  • 输入:{"question": "问题内容", ...}
  • 输出:{"question": "问题内容", "context": "文档1:xxxx...\\n\\n文档2:xxxx..." ...}
context_gen = RunnablePassthrough.assign(context=get_context)

创建提示词

prompt = ChatPromptTemplate.from_messages([("system", "你是一个友好的助手,这是{context}。"),MessagesPlaceholder(variable_name="history"),("human", "{question}")
])

初始化大模型

llm = ChatTongyi(model_name="qwen-max",dashscope_api_key=key,streaming=True
)

创建基础处理链

base_chain = context_gen | prompt | llm

创建全局存储字典

  • :会话ID(session_id)
  • InMemoryHistory 实例
store = {}

实现内存历史存储类

class InMemoryHistory(BaseChatMessageHistory, BaseModel):messages: List[BaseMessage] = Field(default_factory=list)def add_messages(self, messages: List[BaseMessage]) -> None:self.messages.extend(messages)def clear(self) -> None:self.messages = []

获取会话历史

def get_session_history(session_id):if session_id not in store:store[session_id] = InMemoryHistory()return store[session_id]

创建带有历史的对话链

chain_with_history = RunnableWithMessageHistory(base_chain,get_session_history,input_messages_key="question",history_messages_key="history"
)

测试

r4 = chain_with_history.invoke({"question": "太上老君炼丹炉的起源传说,出自哪里?"},config={'configurable': {"session_id": "test"}}
)
print(r4)r5 = chain_with_history.invoke({"question": "我刚才问的什么?"},config={'configurable': {"session_id": "test"}}
)
print(r5)

http://www.dtcms.com/a/272789.html

相关文章:

  • Transformers 和 PyTorch 的区别与安装指南
  • Docker 高级管理--Dockerfile镜像制作
  • Context Engineering Framework 系统详细介绍
  • 链表算法之【合并两个有序链表】
  • 牛客笔试题 除2!
  • 读取按键的四种方式
  • IMU误差模型
  • 显卡GPU的架构和工作原理
  • 输入框过滤选项列表,el-checkbox-group单选
  • JDK 1.7 vs JDK 1.8
  • 为什么域名加端口访问需要放行端口?
  • 【算法训练营Day11】二叉树part1
  • c语言初阶 指针
  • CH9121T电路及配置详解
  • 【算法笔记 day three】滑动窗口(其他类型)
  • Spring Security 技术原理与实战全景详解
  • 【OD机试题解法笔记】根据IP查找城市
  • 观众信息设置与统计(视频高级分析与统计功能)
  • 身份认证缺陷
  • Gulp实现功能及插件总结
  • java并发包下CountDownLatch、Semaphore用法
  • 【牛客刷题】活动安排
  • i.mx8 网络速率测试
  • Transformer:自注意力驱动的神经网络革命引擎
  • 网络综合实验
  • Linux中gdb使用
  • Spring- @Autowired和@Resource 的区别
  • 【网络】Linux 内核优化实战 - net.ipv4.tcp_keepalive_probes
  • linux系统---部署应用
  • day049-初识Ansible与常用模块