1. 项目概述：为什么选择LangChain与通义千问？

最近在折腾AI应用开发的朋友，估计没少被“如何让大模型干点实际的、自动化的事情”这个问题困扰。直接调用模型API，写个简单的对话还行，一旦涉及到多步骤推理、工具调用、记忆管理，代码复杂度就直线上升，自己从头造轮子既费时又容易出bug。这正是LangChain这类框架的价值所在——它提供了一套标准化的“积木”，让我们能像搭乐高一样，快速构建起功能复杂的智能体（Agent）。

而通义千问，作为国内顶尖的大语言模型之一，其强大的理解、推理和代码能力，让它成为构建智能体应用的绝佳“大脑”。但官方SDK通常只提供基础的对话接口，如何将其无缝接入LangChain的生态，赋予它使用工具、访问网络、处理文档等“手脚”，就成了一个关键的技术节点。

这个项目，就是一次从零开始的实战记录。我将手把手带你完成LangChain与通义千问的深度集成，不仅仅是简单的API调用，而是构建一个具备基础Agent能力的应用骨架。你会看到如何配置环境、封装模型、处理流式输出，并初步探索工具调用的接口。无论你是想做一个自动化的数据分析助手，还是一个能联网查询的智能客服原型，这里面的步骤都是通用的基础。我们避开那些华而不实的理论，直接进入可以运行、可以修改、可以扩展的代码层面。

2. 环境准备与核心依赖解析

动手之前，先把“厨房”收拾好。这里的环境配置不仅仅是安装几个包，更重要的是理解每个包的作用以及版本兼容性背后的逻辑，这能帮你避开未来无数个令人头疼的依赖冲突问题。

2.1 Python环境与虚拟隔离

首先，我强烈建议使用Python 3.9或3.10。Python 3.11+在某些科学计算库上可能仍有兼容性问题，而3.8又略显老旧。3.9和3.10是目前最稳定、生态支持最全面的版本，能最大程度减少不必要的麻烦。

其次，务必使用虚拟环境。这是Python项目管理的黄金法则。它能为你的项目创建一个独立的Python包安装空间，与系统环境和其他项目完全隔离。

# 使用venv创建虚拟环境（Python 3.3+内置） python -m venv venv_langchain_qwen # 激活虚拟环境 # 在Windows上： venv_langchain_qwen\Scripts\activate # 在macOS/Linux上： source venv_langchain_qwen/bin/activate

激活后，你的命令行提示符前通常会显示(venv_langchain_qwen)，表示你已经进入了这个独立环境。后续所有pip install操作都只影响这个环境。

注意：很多初学者会忽略这一步，直接把包装在全局环境里。一旦项目多了，不同项目对同一个包有不同版本要求，就会引发难以排查的冲突。虚拟环境是专业开发的起点。

2.2 依赖包安装与选型理由

接下来安装核心依赖。我们使用pip进行安装。请将以下内容保存为requirements.txt文件，然后执行pip install -r requirements.txt。

langchain==0.1.0 langchain-community==0.0.10 openai==1.12.0 httpx==0.26.0 pydantic==2.5.0 python-dotenv==1.0.0

现在，我来逐一解释为什么是这些包以及它们的版本：

1. langchain==0.1.0: 这是LangChain的核心库。选择0.1.0这个相对较新的稳定版本，是因为LangChain近期进行了重大版本重构（从0.0.x到0.1.x），API变化较大。许多老教程的代码在新版本上已无法运行。我们直接使用新版本的范式，避免走弯路。
2. langchain-community==0.0.10: 在LangChain新版本中，许多第三方集成（包括社区贡献的模型封装、工具等）被剥离到了这个包。我们要调用通义千问，就需要它里面提供的ChatTongyi类。
3. openai==1.12.0: 这看起来有点奇怪，我们不是用通义千问吗？没错，但LangChain设计了一套通用的ChatModel接口，其最初是围绕OpenAI的API格式设计的。通义千问的官方API在参数和响应格式上与OpenAI API高度兼容。安装这个包，是为了利用其定义的OpenAI兼容的客户端和基础类，ChatTongyi会继承和适配它。这是实现兼容性的关键。
4. httpx==0.26.0: 一个现代、快速且功能丰富的HTTP客户端库。openai库底层使用它来发起网络请求。指定版本是为了确保与openai库的兼容性。
5. pydantic==2.5.0: 一个数据验证和设置管理库，LangChain大量使用它来定义模型输入输出的数据结构（Schema）。V2版本性能提升巨大，且LangChain新版本已适配。
6. python-dotenv==1.0.0: 用于从.env文件加载环境变量（如API Key）。将敏感信息与代码分离是基本的安全实践。

安装完成后，可以通过pip list命令检查是否安装成功。

2.3 获取通义千问API-KEY

一切就绪，只欠东风——通义千问的API访问凭证。

1. 访问阿里云官网，注册并登录。在控制台找到“灵积模型服务”（ModelScope）或直接搜索“通义千问”。
2. 开通“通义千问”相关API服务（例如qwen-max、qwen-plus或qwen-turbo）。通常新用户会有免费额度。
3. 在API密钥管理页面，创建一个新的API Key。你会得到类似sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx的一串密钥。请立即妥善保存，它只显示一次。

为了安全，我们绝不将API Key硬编码在代码中。在项目根目录创建一个名为.env的文件，内容如下：

DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

这里的DASHSCOPE_API_KEY是通义千问API约定的环境变量名。python-dotenv库会帮我们自动加载它。

实操心得：免费额度虽然够用，但最好在阿里云控制台设置一下“额度报警”，避免意外超支。另外，不同模型（如qwen-max和qwen-turbo）的计费标准和能力不同，初期测试可以用qwen-turbo，成本低、响应快。

3. 核心实现：封装通义千问为LangChain ChatModel

环境备好，钥匙在手，现在我们来打造连接LangChain和通义千问的“桥梁”。这一步是整个项目的核心，理解了这里的封装逻辑，你就能举一反三，接入其他兼容OpenAI API的模型。

3.1 创建模型封装文件

在项目目录下，我们创建一个Python文件，例如qwen_langchain.py。这个文件将包含我们自定义的模型封装类。

首先，导入必要的模块：

import os from typing import Any, Dict, List, Optional, Union, Iterator from langchain_core.callbacks import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage from langchain_core.outputs import ChatGeneration, ChatResult from pydantic import Field, SecretStr from openai import OpenAI

关键点解析：

• BaseChatModel: 这是LangChain所有聊天模型的基类。我们的类需要继承它，并实现几个关键方法。
• BaseMessage及其子类：这是LangChain中消息的通用格式。SystemMessage、HumanMessage、AIMessage分别对应系统提示、用户输入和AI回复。
• ChatResult: 模型调用返回的标准结果格式。
• OpenAI: 来自openai包，我们利用它兼容的客户端来发送请求。

3.2 实现ChatTongyi类

接下来，我们实现核心的ChatTongyi类。虽然langchain-community中可能已有类似实现，但自己实现一遍能让你彻底理解其工作原理。

class ChatTongyi(BaseChatModel): """通义千问模型的LangChain封装。""" # 模型名称，例如 qwen-max, qwen-plus, qwen-turbo model_name: str = Field(default="qwen-max", alias="model") # API密钥，使用SecretStr进行安全封装 dashscope_api_key: SecretStr # API基础URL，指向通义千问的端点 base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1" # 温度参数，控制随机性 temperature: float = 0.8 # 客户端实例，内部使用 _client: Optional[OpenAI] = None class Config: """Pydantic配置，允许通过别名设置字段。""" allow_population_by_field_name = True def __init__(self, **kwargs): super().__init__(**kwargs) # 初始化OpenAI兼容客户端 self._client = OpenAI( api_key=self.dashscope_api_key.get_secret_value(), base_url=self.base_url, timeout=60.0, # 设置超时时间 ) @property def _llm_type(self) -> str: """返回模型类型标识，用于LangChain内部记录。""" return "tongyi-chat" def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRRun] = None, **kwargs: Any, ) -> ChatResult: """核心生成方法，将LangChain消息格式转换为通义千问API请求。""" # 1. 将LangChain消息格式转换为OpenAI API格式 openai_messages = [] for msg in messages: if isinstance(msg, SystemMessage): role = "system" elif isinstance(msg, HumanMessage): role = "user" elif isinstance(msg, AIMessage): role = "assistant" else: # 处理其他可能的消息类型 role = getattr(msg, "type", "user") openai_messages.append({ "role": role, "content": msg.content }) # 2. 构建请求参数 params = { "model": self.model_name, "messages": openai_messages, "temperature": self.temperature, **kwargs } if stop: # 通义千问API可能使用`stop_sequences`或其他参数，这里需要适配 # 注意：通义千问的停止词参数可能与OpenAI不完全一致，需查阅最新文档 params["stop"] = stop # 3. 调用API try: response = self._client.chat.completions.create(**params) except Exception as e: # 处理网络错误、认证错误、额度不足等异常 raise ValueError(f"调用通义千问API失败: {e}") # 4. 将API响应转换回LangChain的ChatResult格式 message = AIMessage(content=response.choices[0].message.content) generation = ChatGeneration(message=message) return ChatResult(generations=[generation]) # 实现流式输出方法（可选但重要） def _stream( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> Iterator[ChatGenerationChunk]: """流式生成方法，用于实时输出token。""" # 转换消息格式（同上） openai_messages = [] for msg in messages: if isinstance(msg, SystemMessage): role = "system" elif isinstance(msg, HumanMessage): role = "user" elif isinstance(msg, AIMessage): role = "assistant" else: role = getattr(msg, "type", "user") openai_messages.append({"role": role, "content": msg.content}) params = { "model": self.model_name, "messages": openai_messages, "temperature": self.temperature, "stream": True, # 开启流式 **kwargs } if stop: params["stop"] = stop stream = self._client.chat.completions.create(**params) for chunk in stream: if chunk.choices and chunk.choices[0].delta.content is not None: content = chunk.choices[0].delta.content # 将每个token块封装为ChatGenerationChunk yield ChatGenerationChunk(message=AIMessageChunk(content=content)) # 如果提供了回调管理器，可以触发流式回调 if run_manager: run_manager.on_llm_new_token(content)

这个类做了以下几件关键事：

1. 继承与配置：继承BaseChatModel，并定义了模型必需的配置参数（model_name,api_key等）。
2. 初始化客户端：在__init__中，使用通义千问的API Key和端点URL，初始化了一个OpenAI兼容的客户端。这是能成功调用的技术关键。
3. 实现_generate方法：这是强制要求的方法。它负责：
   • 格式转换：把LangChain通用的BaseMessage列表，转换成通义千问API能识别的role和content格式的字典列表。
   • 参数组装：合并温度、停止词等参数。
   • 发起请求与异常处理：调用客户端接口，并包裹在try-catch中，提供友好的错误提示。
   • 结果封装：将API返回的content，重新封装成LangChain的AIMessage和ChatResult对象。
4. 实现_stream方法（可选）：为了让模型支持流式输出（一个字一个字往外蹦的效果），我们需要实现这个方法。其逻辑与_generate类似，但需要处理stream=True参数并迭代响应流。

注意事项：通义千问的API参数和OpenAI并非100%一致。例如，停止词参数、最大token数参数名可能需要调整。上述代码是一个通用模板，在实际使用中，你需要根据 通义千问官方API文档 的最新说明，微调params字典中的键名。这是集成第三方模型最常见的适配工作。

3.3 编写使用示例与测试

封装完成后，我们写一个简单的脚本来测试它。创建一个test_qwen.py文件：

import os from dotenv import load_dotenv from qwen_langchain import ChatTongyi from langchain_core.messages import HumanMessage, SystemMessage # 1. 加载环境变量 load_dotenv() api_key = os.getenv("DASHSCOPE_API_KEY") if not api_key: raise ValueError("请在 .env 文件中设置 DASHSCOPE_API_KEY") # 2. 初始化模型 llm = ChatTongyi( dashscope_api_key=api_key, model_name="qwen-turbo", # 使用turbo模型测试，更快更便宜 temperature=0.1, # 低温度，输出更确定 ) # 3. 构建消息链 messages = [ SystemMessage(content="你是一个乐于助人的AI助手，回答要简洁准确。"), HumanMessage(content="请用Python写一个函数，计算斐波那契数列的第n项。") ] # 4. 调用模型（非流式） print("=== 非流式调用 ===") try: response = llm.invoke(messages) print(response.content) except Exception as e: print(f"调用出错: {e}") # 5. 调用模型（流式） print("\n=== 流式调用 ===") try: for chunk in llm.stream(messages): print(chunk.content, end="", flush=True) print() # 换行 except Exception as e: print(f"\n流式调用出错: {e}")

运行这个脚本python test_qwen.py，如果一切配置正确，你应该会先看到一段完整的Python代码输出，然后以流式方式再输出一遍。这证明我们的封装成功了，模型已经可以正常通过LangChain接口进行对话。

4. 迈向智能体：基础工具调用与链的构建

仅仅能对话还不够，智能体的核心在于能“使用工具”。接下来，我们为这个通义千问模型装上第一个“工具”，并体验LangChain最核心的“链”式编排。

4.1 创建一个简单的计算器工具

在LangChain中，工具是一个可以被模型调用的函数。我们先创建一个最简单的工具——一个能进行四则运算的计算器。

在qwen_langchain.py同目录下，创建tools.py：

from langchain.tools import tool from math import sqrt, log, sin, cos, tan, pi # 引入更多数学函数 @tool def calculator(expression: str) -> str: """ 一个安全的计算器工具。输入一个数学表达式字符串，返回计算结果。 支持加减乘除（+, -, *, /）、乘方（**）、括号和常见数学函数。 例如: `(3 + 5) * 2 / 4`, `sqrt(16)`, `sin(pi/2)`。 Args: expression: 数学表达式字符串。 Returns: 计算结果的字符串，或错误信息。 """ # 安全警告：直接使用eval是极度危险的，因为它可以执行任意代码。 # 这里仅作为演示，在实际生产环境中，必须使用安全的表达式求值库，如 `asteval`。 # 我们这里做一个极简的安全过滤（仅用于演示，不保证绝对安全）。 allowed_chars = set("0123456789+-*/.() sqrtlogsin costanpi **") if not all(c in allowed_chars for c in expression.replace(' ', '')): return "错误：表达式中包含不安全字符。" try: # 将**替换为pow，并求值（仍然不安全，仅演示用） # 生产环境请务必使用 asteval 或类似库！ result = eval(expression, {"__builtins__": {}}, {"sqrt": sqrt, "log": log, "sin": sin, "cos": cos, "tan": tan, "pi": pi}) return str(result) except Exception as e: return f"计算错误: {e}" # 可以定义更多工具 @tool def get_current_time(placeholder: str = "") -> str: """获取当前的日期和时间。输入参数可以忽略。""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

重要安全提示：上面的calculator工具为了演示简便，使用了eval，这在真实项目中是严重的安全漏洞，攻击者可以通过它执行系统命令。在实际应用中，你必须使用安全的表达式求值库，例如asteval，它提供了一个没有危险内置函数的命名空间。这里我们仅用于本地测试和原理演示。

4.2 将工具绑定到模型，创建智能体

有了工具，我们需要让模型知道它有哪些工具可用，并学会在合适的时候调用它们。这需要用到LangChain的bind_tools方法和Agent相关的执行器。

创建agent_demo.py：

import os from dotenv import load_dotenv from qwen_langchain import ChatTongyi from tools import calculator, get_current_time from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate # 加载环境与模型 load_dotenv() api_key = os.getenv("DASHSCOPE_API_KEY") llm = ChatTongyi(dashscope_api_key=api_key, model_name="qwen-turbo", temperature=0) # 1. 定义工具列表 tools = [calculator, get_current_time] # 2. 创建提示模板 # 这是引导模型使用工具的关键。LangChain有预定义的模板，我们也可以自定义。 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个强大的助手，可以调用工具来解决问题。请根据用户问题，决定是否需要调用工具以及调用哪个工具。如果你决定调用工具，必须严格按照要求的格式输出。"), ("placeholder", "{chat_history}"), # 预留对话历史的位置 ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), # 代理的“思考草稿纸”，用于记录工具调用和结果 ]) # 3. 创建工具调用智能体 agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt) # 4. 创建代理执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # 5. 运行测试 print("=== 智能体测试开始 ===") questions = [ "123乘以456等于多少？", "现在几点了？", "先计算125的平方根，再加上当前时间的分钟数，最后结果是多少？", # 这是一个需要多步推理和多次调用工具的问题 ] for question in questions: print(f"\n用户: {question}") try: # 注意：这里我们传入空的chat_history，因为这是单轮对话 result = agent_executor.invoke({"input": question, "chat_history": []}) print(f"助手: {result['output']}") except Exception as e: print(f"执行出错: {e}")

运行这个脚本，你会看到类似以下的输出（verbose=True会打印详细过程）：

=== 智能体测试开始 === 用户: 123乘以456等于多少？ > 进入新的AgentExecutor链... 思考：用户需要计算123和456的乘积。我有一个计算器工具可以处理这个。 行动：

{ "action": "calculator", "action_input": "123 * 456" }

观察: 56088 思考：我得到了计算结果，可以回答用户了。 行动：

{ "action": "Final Answer", "action_input": "123乘以456等于56088。" }

> 链结束。 助手: 123乘以456等于56088。

你会观察到，模型并没有直接输出答案，而是先“思考”是否需要调用工具，然后输出了一个结构化的JSON，指定要调用的工具名称和输入参数。AgentExecutor捕获到这个JSON，就去执行对应的calculator函数，并将执行结果“观察: 56088”返回给模型。模型再根据这个观察结果，生成最终的回答。

这就是LangChain智能体最基础的工作流：规划（Plan）-> 执行（Act）-> 观察（Observe）-> 再规划（Plan）...直到任务完成。

4.3 解析工具调用流程与关键参数

上面的例子展示了最简单的工具调用。我们来深入拆解几个关键点：

1. bind_tools与create_tool_calling_agent：新版本的LangChain推荐使用create_tool_calling_agent来快速创建基于OpenAI Function Calling格式的智能体。它内部会帮我们做两件事：一是将工具的描述信息“绑定”到LLM（让LLM知道有哪些工具及其功能），二是设置好提示模板，引导LLM以特定格式输出工具调用请求。
2. 提示模板（PromptTemplate）：系统提示词至关重要。它定义了模型的角色和行为规范。{agent_scratchpad}是一个特殊的占位符，执行器会自动将模型的思考过程、工具调用和工具结果填充到这里，供模型在下一轮推理时参考。这就是实现多步推理的“记忆”机制。
3. AgentExecutor：这是智能体的“发动机”。它负责：
   • 解析模型输出的文本，识别其中的工具调用指令（通常是JSON）。
   • 根据指令找到对应的工具函数并执行。
   • 将工具执行的结果格式化后，连同历史对话一起，再次喂给模型。
   • 循环这个过程，直到模型输出最终答案（Final Answer）。
   • handle_parsing_errors=True参数非常有用，当模型输出的JSON格式不对时，它会尝试让模型重试，而不是直接崩溃。
4. 流式输出与智能体：目前的AgentExecutor默认不支持将模型的流式输出（一个字一个字）展示给用户，因为它内部需要进行多轮推理。如果你需要流式体验，通常是在最终答案生成阶段，或者需要使用更底层的Runnable接口进行自定义编排。

实操心得：在测试工具调用时，一开始失败率可能不低。常见问题有：1）模型不按指定格式输出JSON。这需要优化你的系统提示词。2）工具描述不够清晰。确保@tool装饰器下的文档字符串（docstring）清晰、准确地描述了工具的功能和输入参数格式，模型主要靠这个来理解工具。3）温度（temperature）设置过高。对于工具调用这类需要严格遵循格式的任务，将temperature设为0或接近0的值，能显著提高输出的稳定性和准确性。

5. 常见问题、排查技巧与性能优化

在实际集成和开发过程中，你一定会遇到各种问题。下面是我踩过坑后总结的一些常见问题及其解决方案，以及一些提升应用性能的实用技巧。

5.1 集成与调用问题排查表

5.2 性能优化与最佳实践

当你的智能体应用跑起来后，下一步就是让它跑得更快、更稳、更省钱。

1. 模型选型与成本控制：

   • 场景匹配：qwen-max能力最强但最贵且慢，适合复杂推理、创作。qwen-turbo响应快、成本低，适合简单问答、分类、提取。根据任务复杂度选择合适的模型。
   • 缓存：对重复或相似的问题，使用LangChain的缓存功能（如InMemoryCache,SQLiteCache）可以避免重复调用API，大幅节省成本和时间。

   from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache set_llm_cache(InMemoryCache())

2. 超时与重试：

   • 网络不稳定或API偶发性拥堵会导致请求超时。在初始化客户端或模型时，务必设置合理的超时时间（如timeout=30.0）。
   • 使用LangChain的RunnableWithRetry或自定义重试逻辑，对可重试的错误（如网络超时、5xx服务器错误）进行有限次数的重试。

3. 提示工程优化：

   • 结构化输出：对于需要模型返回特定格式（如JSON、列表）的场景，在系统提示词中明确要求，并给出示例（Few-Shot Prompting），可以极大提高输出格式的准确性。
   • 分步思考（Chain-of-Thought）：对于复杂问题，在提示词中要求模型“让我们一步步思考”，可以提升其推理能力和工具调用的准确性。

4. 异步调用：

   • 如果你的应用需要同时处理多个用户请求或调用多个工具，使用异步（Async）接口可以显著提高吞吐量。LangChain的很多组件都支持ainvoke,astream,abatch等异步方法。

   import asyncio async def async_call(): result = await agent_executor.ainvoke({"input": "你好", "chat_history": []}) print(result) asyncio.run(async_call())

5. 日志与监控：

   • 在生产环境中，记录每一次API调用的输入、输出、耗时和token使用量至关重要。这有助于分析成本、优化提示词、排查问题。你可以使用LangChain的Callback（回调）机制，轻松地将这些信息输出到文件、数据库或监控系统。

5.3 从Demo到生产：安全与扩展考量

我们上面的Demo为了简洁，省略了很多生产环境必需的环节：

• 工具安全：如前所述，绝对不要在生产环境中使用eval。使用asteval、numexpr或自己编写安全的解析器。
• 错误处理与降级：智能体可能陷入死循环、调用不存在工具或参数错误。AgentExecutor的max_iterations参数可以限制最大循环次数，防止死循环。你需要设计兜底策略，比如在多次失败后，让模型直接以自然语言回答“我无法完成这个任务”。
• 记忆管理：我们的例子使用了单轮对话（空chat_history）。真实的聊天助手需要记忆历史。LangChain提供了多种记忆后端，如ConversationBufferMemory、ConversationSummaryMemory等，你需要将它们集成到AgentExecutor的输入中。
• 扩展更多工具：智能体的能力取决于工具集。你可以封装任何API或函数作为工具，例如：搜索网页、查询数据库、发送邮件、操作文件等。langchain-community包中已经提供了大量现成工具（如SerpAPIWrapper,WikipediaQueryRun），可以直接使用。

走到这一步，你已经拥有了一个由通义千问驱动、具备基础工具调用能力的LangChain智能体骨架。它就像一辆装上了引擎和方向盘的汽车，虽然内饰还很简陋，但已经可以跑起来了。接下来的工作，就是为它添加更多的功能模块（工具）、更舒适的交互体验（记忆、流式）以及更坚固的安全外壳，让它能够胜任真实的场景。这个过程会不断遇到新的挑战，但每一次解决问题的过程，都是你对AI应用开发生态理解加深的时刻。