如何用Python LangChain十分钟玩转阿里通义千问？

本文全程基于 Python + LangChain 框架，带你从0 到 1搭建基于阿里通义千问大模型的应用，从最简单的一次性对话，逐步进阶到循环对话、上下文记忆、工具调用、企业级知识库，完整覆盖大模型应用的核心能力。 前置准备 环境配置 安装依赖包 pip install langchain langchain-community langchain-core python-dotenv 获取阿里通义千问 API Key 登录阿里云百炼平台 创建 API-KEY，保存备用 项目根目录创建 .env 文件，存放密钥： DASHSCOPE_API_KEY=你的阿里通义千问API_KEY 一、基础入门：一次性对话（无记忆、无交互） 这是大模型应用的最小单元，仅实现单次提问、单次回答，无任何额外能力。 核心代码 from langchain_community.chat_models import ChatTongyi from dotenv import load_dotenv from langchain_core.messages import HumanMessage load_dotenv() llm = ChatTongyi() message = [HumanMessage(content="介绍下你自己")] print(llm.invoke(message).content) 测试效果 核心知识点 ChatTongyi：langchain_community 封装的阿里通义千问对话大模型调用类 invoke()：同步调用大模型的核心方法 适用场景：简单的一次性问答、批量文本处理 二、进阶交互：循环对话（控制台持续聊天） 在基础上实现控制台持续对话，用户可以反复提问，直到主动退出。 核心代码 from langchain_community.chat_models import ChatTongyi from dotenv import load_dotenv from langchain_core.messages import HumanMessage load_dotenv() llm = ChatTongyi() print("循环对话: 输入exit推出！\n") while True: user_input = input("你：") if user_input == "exit": break message = [HumanMessage(content=user_input)] aimessage = llm.invoke(message) print(f"AI:{aimessage.content}\n") 测试效果 核心知识点 实现了交互式终端，贴近真实聊天工具的使用体验 缺陷：无记忆能力，模型不知道你上一轮说了什么（上次对话告诉了大模型我的名字，下次对话再问它它就不知道了） 三、核心能力：上下文记忆（记住历史对话） 大模型应用的核心能力：让模型记住多轮对话历史，实现连贯聊天。 简版上下文记忆 核心代码（记录每次对话 + 请求大模型时将所有历史会话扔给大模型） from langchain_community.chat_models import ChatTongyi from dotenv import load_dotenv from langchain_core.messages import HumanMessage, AIMessage load_dotenv() llm = ChatTongyi() messages = [] print("循环对话(带上下文记忆): 输入exit推出！\n") while True: user_input = input("你：") if user_input == "exit": break messages.append(HumanMessage(content=user_input)) aimessage = llm.invoke(messages) messages.append(AIMessage(content=aimessage.content)) print(f"AI:{aimessage.content}\n") 测试效果 核心知识点 messages：简化板记忆组件，就是一个数组，存储完整对话历史（输入信息HumanMessage，AI的回复AIMessage） 后面我们会再使用LangChain中封装的记忆类型扩展以及对话连（可以实时总结记忆，节省token） 四、高级扩展：工具调用（让大模型会查天气、查时间） 大模型本身无法联网、无法计算，通过工具调用赋予模型实时能力。 示例：给大模型添加「查询时间、获取天气」 import datetime import requests from dotenv import load_dotenv from langchain_community.chat_models import ChatTongyi from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool load_dotenv() # ===== 1. 定义工具函数 ===== @tool def get_current_time() -> str: """获取当前日期和时间""" return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") @tool def get_weather(city): """获取天气信息""" url = f"https://uapis.cn/api/v1/misc/weather?city={city}" response = requests.get(url) data = response.json() return data tools = [get_current_time, get_weather] # ===== 2. 初始化 LLM 并绑定工具 ===== llm = ChatTongyi(model="tongyi-xiaomi-analysis-pro", temperature=0.7) llm_with_tools = llm.bind_tools(tools) # 👈 关键：启用工具调用能力 # ===== 3. 手动维护对话历史 ===== messages = [] # 扁平消息列表：HumanMessage / AIMessage / ToolMessage print("欢迎使用带工具的 AI 助手！输入 exit 退出：") while True: user_input = input("你: ") if user_input == "exit": break # 添加用户消息 messages.append(HumanMessage(content=user_input)) # 第一次调用：可能返回 tool_calls ai_response = llm_with_tools.invoke(messages) # 如果模型要求调用工具 if hasattr(ai_response, 'tool_calls') and ai_response.tool_calls: # 保存模型的 tool_call 请求（用于上下文） messages.append(ai_response) # 执行所有工具调用 for tool_call in ai_response.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] # 查找对应工具 func = next((t for t in tools if t.name == tool_name), None) if func: result = func.invoke(tool_args) # 执行工具 else: result = f"工具 '{tool_name}' 未找到" # 将工具结果作为 ToolMessage 加入上下文 messages.append(ToolMessage( content=str(result), tool_call_id=tool_call["id"] )) # 再次调用 LLM，让它基于工具结果生成最终回答 ai_response = llm.invoke(messages) # 添加到记忆中 messages.append(ai_response) print(f"AI: {ai_response.content}\n") 测试效果 具体流程 先来看下 messages 变量的值，根据这个值我们可以看到整个对话的流程 流程图： 第一轮对话：查询日期 用户输入：HumanMessage(content='今天的日期是多少') LLM 决策：识别到需要实时信息，生成 AIMessage 并携带 tool_calls，请求调用 get_current_time() 工具 工具执行：执行 get_current_time()，返回 ToolMessage(content='2026-03-29 11:06:05') 生成回答：LLM 结合工具结果，生成最终 AIMessage(content='2026-03-29') 第二轮对话：查询天气（上下文延续） 用户输入：HumanMessage(content='今天北京的天气怎么样')（基于上一轮对话的上下文） LLM 决策：识别到需要外部天气数据，生成 AIMessage 并携带 tool_calls，请求调用 get_weather(city="北京") 工具执行：执行 get_weather("北京")，返回包含详细天气数据的 ToolMessage 生成回答：LLM 整理工具结果，生成最终 AIMessage 自然语言回答 🧠 核心机制说明 上下文记忆：所有 HumanMessage / AIMessage / ToolMessage 都被追加到 messages 列表中，LLM 每次调用都基于完整历史上下文进行推理 工具调用流程： LLM 判断是否需要工具 生成结构化 tool_calls 指令 执行工具并返回 ToolMessage LLM 结合工具结果生成最终回答 消息类型： HumanMessage：用户输入 AIMessage：AI 回复或工具调用指令 ToolMessage：工具执行结果 五、企业级应用：私有知识库问答（RAG） 大模型的终极落地形态：基于私有文档回答问题，不泄露数据，不依赖模型固有知识。 新增依赖 pip install langchain-text-splitters faiss-cpu pypdf 核心代码（RAG 知识库） import datetime import requests from dotenv import load_dotenv from langchain_community.chat_models import ChatTongyi from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.documents import Document from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from langchain_text_splitters import RecursiveCharacterTextSplitter load_dotenv() # ===== 1. 构建向量知识库 ===== class VectorKnowledgeBase: """基于 FAISS 的向量知识库""" def __init__(self, embedding_model="multimodal-embedding-v1"): self.embedding_model = DashScopeEmbeddings(model=embedding_model) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, length_function=len ) self.vectorstore = None self.documents = [] # 初始化知识库数据 self._load_mock_data() self._build_vectorstore() def _load_mock_data(self): """加载模拟数据""" mock_data = [ # 公司信息 Document( page_content="未来科技有限公司成立于 2020 年，位于北京市海淀区中关村，是一家专注于人工智能、云计算、大数据分析的高新技术企业。公司现有员工 500 人，CEO 是张三先生。", metadata={"category": "company", "type": "overview"} ), Document( page_content="公司联系方式：邮箱 contact@futuretech.com，电话 400-888-9999，地址：北京市海淀区中关村科技园 88 号。", metadata={"category": "company", "type": "contact"} ), # 产品信息 Document( page_content="智能客服系统 v3.0：基于大模型的智能客服解决方案，支持自然语言理解、多轮对话、情感分析等功能，可帮助企业提升客服效率 80%。价格：¥99,999/年。", metadata={"category": "products", "type": "customer_service"} ), Document( page_content="数据分析平台 v2.5：企业级大数据分析平台，支持实时数据处理、可视化报表、机器学习建模等功能，已服务超过 1000 家企业客户。价格：¥199,999/年。", metadata={"category": "products", "type": "analytics"} ), Document( page_content="云存储系统 v1.8：安全可靠的云存储解决方案，提供 99.99% 的数据持久性，支持自动备份、版本控制、权限管理等功能。价格：¥49,999/年起。", metadata={"category": "products", "type": "cloud_storage"} ), # 政策制度 Document( page_content="工作时间制度：工作日为周一至周五，上班时间为 7:00-19:00，中午休息 1 小时。弹性工作制，核心工作时间为 10:00-16:00。", metadata={"category": "policies", "type": "work_hours"} ), Document( page_content="休假政策：员工享有带薪年假 15 天，带薪病假 7 天，以及国家法定节假日。工作满一年后每年增加 1 天年假。", metadata={"category": "policies", "type": "vacation"} ), Document( page_content="福利保障：公司为员工缴纳五险一金（养老保险、医疗保险、失业保险、工伤保险、生育保险、住房公积金），并提供补充商业保险。", metadata={"category": "policies", "type": "insurance"} ), Document( page_content="培训发展：公司提供专业技能培训补贴，每人每年最高 5000 元。支持在职深造和职业资格认证考试。", metadata={"category": "policies", "type": "training"} ), # 常见问题 Document( page_content="密码重置问题：登录系统后，在个人中心点击忘记密码按钮，通过注册邮箱接收验证邮件，点击邮件中的链接即可重置密码。如未收到邮件，请联系客服。", metadata={"category": "faq", "type": "account"} ), Document( page_content="发票申请：在订单详情页点击申请发票，填写开票信息（公司名称、税号等），电子发票将在 3 个工作日内发送到指定邮箱。", metadata={"category": "faq", "type": "invoice"} ), Document( page_content="退款政策：产品购买后 7 天内可申请无理由退款。超过 7 天需提供相关证明材料，根据具体情况处理。退款将在 5-10 个工作日内原路返回。", metadata={"category": "faq", "type": "refund"} ), Document( page_content="技术支持：提供 7x24 小时技术支持服务，可通过在线客服、电话 400-888-9999、邮件 support@futuretech.com 联系我们。", metadata={"category": "faq", "type": "support"} ), ] self.documents = mock_data def _build_vectorstore(self): """构建向量索引""" try: # 分割文档 all_splits = self.text_splitter.split_documents(self.documents) # 创建 FAISS 向量库 self.vectorstore = FAISS.from_documents( documents=all_splits, embedding=self.embedding_model ) print(f"✓ 向量知识库构建完成，共 {len(all_splits)} 个文档片段\n") except Exception as e: print(f"✗ 向量知识库构建失败：{e}") print("⚠ 将使用备用方案（基于关键词搜索）\n") self.vectorstore = None def search(self, query: str, k: int = 3) -> list: """ 向量相似度搜索 :param query: 查询文本 :param k: 返回最相关的 k 条结果 :return: 相关文档列表 """ if not self.vectorstore: return self._keyword_search(query, k) results = self.vectorstore.similarity_search(query, k=k) return results def search_with_score(self, query: str, k: int = 3) -> list: """ 带分数的向量搜索 :param query: 查询文本 :param k: 返回结果数量 :return: (文档，相似度分数) 列表 """ if not self.vectorstore: results = self._keyword_search(query, k) # 转换为带分数的格式 (关键词搜索没有分数，默认为 0) return [(doc, 0.0) for doc in results] results = self.vectorstore.similarity_search_with_score(query, k=k) return results def _keyword_search(self, query: str, k: int = 3) -> list: """ 备用方案：基于关键词的搜索 :param query: 查询文本 :param k: 返回结果数量 :return: 相关文档列表 """ results = [] query_words = set(query.lower()) for doc in self.documents: content_words = set(doc.page_content.lower()) # 计算词重叠度 overlap = len(query_words & content_words) if overlap > 0: results.append((doc, overlap)) # 按重叠度排序 results.sort(key=lambda x: x[1], reverse=True) return [doc for doc, _ in results[:k]] def add_document(self, content: str, metadata: dict = None): """添加新文档到知识库""" if not self.vectorstore: print("⚠ 向量库未初始化，无法添加文档") return doc = Document(page_content=content, metadata=metadata or {}) splits = self.text_splitter.split_documents([doc]) self.vectorstore.add_documents(splits) print(f"✓ 已添加新文档，当前共 {len(self.vectorstore.docstore._dict)} 个文档片段") # 初始化向量知识库 vector_kb = VectorKnowledgeBase() # ===== 2. 定义工具函数 ===== @tool def get_current_time() -> str: """获取当前日期和时间""" return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") @tool def get_weather(city: str): """获取天气信息""" url = f"https://uapis.cn/api/v1/misc/weather?city={city}" response = requests.get(url) data = response.json() return data @tool def search_knowledge_base(query: str): """ 使用向量搜索查询私有知识库 :param query: 搜索问题或关键词 """ results = vector_kb.search_with_score(query, k=3) if not results: return "未在知识库中找到相关信息" # 格式化结果 formatted = "" for i, (doc, score) in enumerate(results, 1): formatted += f"{i}. {doc.page_content}\n" # 添加元数据信息 if doc.metadata: category = doc.metadata.get("category", "unknown") doc_type = doc.metadata.get("type", "unknown") formatted += f"(分类：{category}, 类型：{doc_type})\n\n" return formatted if formatted else "未找到相关信息" @tool def get_weather_forecast(city: str): """获取天气预报""" url = f"https://uapis.cn/api/v1/misc/weather?city={city}" response = requests.get(url) data = response.json() return data tools = [get_current_time, get_weather, search_knowledge_base, get_weather_forecast] # ===== 3. 初始化 LLM 并绑定工具 ===== llm = ChatTongyi(model="tongyi-xiaomi-analysis-pro", temperature=0.7) llm_with_tools = llm.bind_tools(tools) # ===== 4. 手动维护对话历史 ===== messages = [] print("🚀 欢迎使用基于 FAISS 向量知识库的 AI 助手！") print("💡 我可以回答关于公司信息、产品、政策等问题") print("输入 exit 退出\n") while True: user_input = input("你：") if user_input == "exit": break messages.append(HumanMessage(content=user_input)) ai_response = llm_with_tools.invoke(messages) if hasattr(ai_response, 'tool_calls') and ai_response.tool_calls: messages.append(ai_response) for tool_call in ai_response.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] func = next((t for t in tools if t.name == tool_name), None) if func: result = func.invoke(tool_args) else: result = f"工具 '{tool_name}' 未找到" messages.append(ToolMessage( content=str(result), tool_call_id=tool_call["id"] )) ai_response = llm.invoke(messages) messages.append(ai_response) print(f"AI: {ai_response.content}\n") 测试结果 核心知识点 RAG（检索增强生成）：大模型应用落地标准方案 流程：文档加载 → 文本分块 → 向量嵌入 → 向量存储 → 检索问答 适用场景：企业文档助手、产品手册、学术论文问答 总结 这篇博客带你完成了大模型应用的完整进化路径： 基础：一次性调用阿里通义千问大模型 交互：实现控制台循环对话 核心：添加上下文记忆，实现连贯聊天 扩展：通过 FunctionCall 调用工具，赋予模型实时能力 落地：基于 RAG 构建私有知识库问答系统 所有代码可直接运行、可直接扩展，是 Python + LangChain 开发阿里大模型应用的最佳入门实践。 关键点回顾 核心依赖：langchain + langchain-community 提供阿里模型封装 记忆能力：定义记忆数组 messages 实现对话上下文留存 工具调用：Tool 让模型具备执行能力 知识库：RAG 是大模型私有化落地的最优解 全程基于阿里通义千问，适配国内生产环境