体验优先：十分钟使用 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 每次调用都基于完整历史上下文进行推理
工具调用流程：
1. LLM 判断是否需要工具
2. 生成结构化 tool_calls 指令
3. 执行工具并返回 ToolMessage
4. 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 是大模型私有化落地的最优解
全程基于阿里通义千问，适配国内生产环境

posted @ 2026-03-29 11:52 wenha 阅读(224) 评论(0) 收藏举报

刷新页面返回顶部

那么远这么近

体验优先：十分钟使用 Python+LangChain 玩转阿里通义千问

前置准备

环境配置

一、基础入门：一次性对话（无记忆、无交互）

核心代码

测试效果

核心知识点

二、进阶交互：循环对话（控制台持续聊天）

核心代码

测试效果

核心知识点

三、核心能力：上下文记忆（记住历史对话）

简版上下文记忆

核心代码（记录每次对话 + 请求大模型时将所有历史会话扔给大模型）

测试效果

核心知识点

四、高级扩展：工具调用（让大模型会查天气、查时间）

示例：给大模型添加「查询时间、获取天气」

测试效果

具体流程

第一轮对话：查询日期

第二轮对话：查询天气（上下文延续）

🧠 核心机制说明

五、企业级应用：私有知识库问答（RAG）

新增依赖

核心代码（RAG 知识库）

测试结果

核心知识点

总结

关键点回顾

公告