内容参考于：图灵AI大模型全栈
上一个内容中写了LCEL链，通过__or__函数实现，在LangChain中它把这个东西封装成了Runable类型，在LangChain我们引入的常用的类都是继承于Runable类型
下方是它的使用示例，用了很多管道的写法
# ====================== 导入依赖库 ======================
# LangChain OpenAI兼容接口客户端：用于调用支持OpenAI API规范的大模型（此处为阿里云通义千问）
from langchain_openai import ChatOpenAI
# 系统消息类：向大模型传递固定角色设定和规则
from langchain_core.messages import SystemMessage
# 聊天提示词模板：结构化构建大模型的输入，支持参数动态填充
from langchain_core.prompts import ChatPromptTemplate
# JSON输出解析器：将大模型的文本输出自动解析为Python字典，保证结构化输出
from langchain_core.output_parsers import JsonOutputParser
# 内存型向量存储：轻量级向量数据库，数据仅存于内存，无需持久化，适合演示和小数据量场景
from langchain_core.vectorstores import InMemoryVectorStore
# LangChain可运行组件：用于构建流式、可组合的AI流程管道
# RunnableMap：并行执行多个任务，统一输入，输出合并为字典
# RunnableBranch：根据条件判断执行不同分支，实现流程分支逻辑
# RunnableLambda：将普通Python函数包装为可运行组件，接入管道
from langchain_core.runnables import RunnableMap, RunnableBranch, RunnableLambda
# Tavily搜索工具：专为AI代理设计的实时搜索引擎，返回结构化、适合LLM处理的网页内容
# 官网：https://www.tavily.com/，支持实时搜索、深度研究、内容提取等功能
from langchain_tavily import TavilySearch
# HuggingFace嵌入模型封装：加载本地或HuggingFace Hub上的文本嵌入模型，将文本转为高维向量
from langchain_huggingface import HuggingFaceEmbeddings
# 加载.env文件工具：从环境变量文件读取敏感信息（API密钥），避免硬编码
from dotenv import load_dotenv
# 操作系统接口：用于读取系统环境变量
import os
# 加载当前目录下.env文件中的环境变量
# .env文件格式示例：
# DASHSCOPE_API_KEY=你的阿里云通义千问API密钥
# TAVILY_API_KEY=你的Tavily搜索API密钥
load_dotenv()
class TravelQASystem:
def __init__(self, openai_api_key, tavily_api_key, embed_path):
"""
初始化旅游问答系统核心组件
入参说明：
1. openai_api_key: str
– 含义：阿里云DashScope平台的API密钥
– 值来源：.env文件中的DASHSCOPE_API_KEY，需在https://dashscope.console.aliyun.com/申请
2. tavily_api_key: str
– 含义：Tavily搜索平台的API密钥（原代码变量名serpapi_api_key为错误命名，已修正）
– 值来源：.env文件中的TAVILY_API_KEY，需在https://www.tavily.com/注册申请
3. embed_path: str
– 含义：文本嵌入模型的本地路径或HuggingFace Hub模型名
– 值来源：用户本地下载的模型文件夹路径（如示例中的BGE中文模型）
– 可选值：也可直接传"HuggingFace Hub模型名"（如"BAAI/bge-large-zh-v1.5"），会自动下载
"""
# ====================== 1. 初始化大语言模型 ======================
# 机制：通过OpenAI兼容接口调用阿里云通义千问
# 阿里云DashScope提供了与OpenAI完全兼容的API端点，因此可直接复用ChatOpenAI类
self.llm = ChatOpenAI(
api_key=openai_api_key, # 通义千问API密钥
# API基础URL：阿里云DashScope OpenAI兼容模式固定端点
# 原文档解析失败但URL本身有效，为官方标准地址
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="qwen-plus" # 调用的模型名称，可选值：qwen-turbo/qwen-plus/qwen-max等
)
# ====================== 2. 初始化实时搜索工具 ======================
# 机制：调用Tavily API获取互联网实时数据（如天气、最新景点信息）
# Tavily优势：专为LLM优化，返回内容已分块、去重、结构化，延迟低（p50仅180ms）
self.search = TavilySearch(tavily_api_key=tavily_api_key)
# ====================== 3. 初始化文本嵌入模型 ======================
# 机制：将文本转换为高维向量，用于后续向量相似度检索
# 示例使用BGE中文大模型，是目前效果最好的中文嵌入模型之一
self.embeddings = HuggingFaceEmbeddings(model_name=embed_path)
# ====================== 4. 构建本地景点知识库 ======================
# 本地静态景点数据，实际项目中可从数据库、CSV文件或爬虫获取
self.attraction_data = [
"故宫：北京地标，明清皇宫，开放时间8:30-17:00",
"颐和园：皇家园林，昆明湖、长廊等景点",
"八达岭长城：距离市区70公里，建议游览3-4小时"
]
# ====================== 5. 初始化内存向量存储 ======================
# 机制：将本地景点文本转换为向量并存储在内存中，实现快速相似度检索
# from_texts方法：自动批量向量化文本并创建向量存储实例
self.vector_store = InMemoryVectorStore.from_texts(
texts=self.attraction_data, # 要向量化的文本列表
embedding=self.embeddings, # 用于向量化的嵌入模型
k=1 # 检索时返回的最相似结果数量，此处只返回最匹配的1条
)
def setup_runnable_pipeline(self):
"""定义LangChain Runnable流程管道，实现问答全流程编排"""
# ====================== 阶段1：问题解析模块 ======================
# 功能：从用户自然语言问题中提取结构化信息（地点、查询类型）
# 提示词模板：明确要求大模型以指定JSON格式输出，保证解析稳定性
parse_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是旅游助手，需从用户问题中提取地点和查询类型（天气/景点介绍/行程规划）"),
("user", """问题：{user_question}请以JSON格式返回：{{"location": "地点", "type": "查询类型"}}""")
])
# 管道串联：提示词模板 → 大模型 → JSON解析器
# 输出：Python字典，格式如{"location": "北京", "type": "天气"}
parse_module = parse_prompt | self.llm | JsonOutputParser()
# ====================== 阶段2：数据获取模块 ======================
# 子模块1：实时天气查询（调用Tavily搜索）
# RunnableLambda：将lambda函数包装为可运行组件，接入管道
weather_query = RunnableLambda(
lambda x: self.search.invoke(f"{x['location']} 今日天气")
)
# 子模块2：本地景点信息检索（向量相似度检索）
# 管道：提取location → 向量检索器 → 提取检索结果的文本内容
# self.vector_store.as_retriever()是Runable类型，Runable类型会前后自动转换所以把(lambda x: x["location"])转换成了RunnableLambda
attraction_retrieval = (lambda x: x["location"]) | self.vector_store.as_retriever() | (
lambda x: x[0].page_content
)
# 并行数据获取：RunnableMap同时执行天气查询和景点检索
# 机制：多个任务并行执行，共享输入，输出合并为字典，提高效率
data_acquisition = RunnableMap({
"weather": weather_query, # 键：weather，值：Tavily返回的天气结果
"attraction": attraction_retrieval, # 键：attraction，值：本地景点文本
"location": (lambda x: x["location"]) # 透传地点信息，供后续使用
})
# ====================== 阶段3：回答生成模块 ======================
# 功能：整合所有获取到的信息，生成自然语言回答
generate_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是专业旅游顾问，需结合景点信息和天气生成建议"),
("user", """地点：{location}
景点信息：{attraction}
天气情况：{weather}
请生成1条行程建议，包含注意事项（如天气相关准备）""")
])
# 管道：提示词模板 → 大模型 → 提取回答内容并去除首尾空白
generate_module = generate_prompt | self.llm | (lambda x: x.content.strip())
# ====================== 阶段4：全流程串联（含条件分支） ======================
# RunnableBranch机制：根据输入条件动态选择执行分支
# 格式：RunnableBranch( (条件函数1, 分支1), (条件函数2, 分支2), 默认分支 )
self.travel_qa_pipeline = (
# 第一步：解析用户问题，得到结构化的location和type
parse_module
# 统一输出格式，为后续分支做准备
| (lambda x: {"location": x["location"], "type": x["type"]})
# 条件分支：根据查询类型选择数据获取路径
| RunnableBranch(
# 分支1：查询类型包含"天气" → 并行获取天气+景点信息
(lambda x: "天气" in x["type"], data_acquisition),
# 默认分支：仅查询景点 → 只获取本地景点信息
lambda x: {"location": x["location"], "attraction": attraction_retrieval.invoke(x)}
)
# 最后一步：生成最终回答
| generate_module
)
def process_user_question(self, user_question):
"""
处理用户提问并返回最终回答
入参：
– user_question: str，用户的自然语言提问
返回：
– str，AI生成的回答内容
"""
input_data = {"user_question": user_question}
# 调用Runnable管道的invoke方法，执行全流程
# 实际项目中建议添加try-except异常处理，捕获API调用失败、解析错误等情况
response = self.travel_qa_pipeline.invoke(input_data)
return response
# 示例用法
if __name__ == "__main__":
# 从环境变量读取API密钥
OPENAI_API_KEY = os.getenv("DASHSCOPE_API_KEY")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY") # 原代码变量名serpapi_api_key已修正
# 本地嵌入模型路径，替换为你自己的实际路径
embed_path = r"D:\\LLM\\Local_model\\BAAI\\bge-large-zh-v1___5"
# 初始化旅游问答系统
travel_qa = TravelQASystem(OPENAI_API_KEY, TAVILY_API_KEY, embed_path)
# 构建Runnable流程管道
travel_qa.setup_runnable_pipeline()
# 测试用例1：查询天气+景点建议（触发并行数据获取分支）
question1 = "今天故宫的天气怎么样?"
answer1 = travel_qa.process_user_question(question1)
print(f"User Question: {question1}\\nAI Answer: {answer1}\\n")
# 测试用例2：仅查询景点介绍（触发默认分支）
# question2 = "颐和园有什么好玩的?"
# answer2 = travel_qa.process_user_question(question2)
# print(f"User Question: {question2}\\nAI Answer: {answer2}\\n")
