在 AI 应用开发的世界里,让大语言模型「动起来」远比让它「说话」更有价值。今天我要分享从零开始使用 LangChain 实现工具调用的完整流程,重点是 MCP(Model Context Protocol)协议的集成方法。

我自己也是从完全不懂 API 的状态开始摸索的,踩过不少坑。这篇教程会用最直白的语言,配合可运行的代码范例,带你一步步做出真正能用的 AI 应用。

什么是工具调用?为什么需要 MCP 协议?

先说说背景。LangChain 是一个帮助开发者快速构建 AI 应用的框架,而「工具调用」(Tool Calling)是让 AI 模型能够执行实际操作的核心功能。传统方式下,开发者需要手动解析模型的输出、再调用对应函数,既繁琐又容易出错。

MCP 协议的出现改变了这个局面。它就像 USB 接口一样——有了统一标准,不同的 AI 模型和不同的工具之间就能无缝对接。你不需要为每个模型写单独的适配代码,只需要按照 MCP 规范定义好工具,框架会自动处理调用逻辑。

在实际项目中,我用 HolySheep AI 作为后端服务,支持 GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash 等多种模型。它的延迟可以控制在 50 毫秒以内,而且费用相比官方渠道能节省 85% 以上,对初学者来说非常友好。

环境准备:从安装到配置

在开始之前,请确保你的电脑已经安装了 Python 3.8 或更高版本。我会假设你已经知道如何打开命令行(Terminal)并运行 Python 脚本。

第一步是安装必要的依赖库。打开终端,输入以下命令:

# 安装 LangChain 核心库和 MCP 扩展
pip install langchain langchain-core langchain-community
pip install mcp

安装 OpenAI 兼容的客户端(HolySheep 使用 OpenAI 格式)

pip install openai

安装 Pydantic 用于数据验证

pip install pydantic

安装完成后,我们需要配置 API 密钥。这是连接 AI 服务的「门票」。你可以创建一个名为 .env 的文件(注意前面有个点),内容如下:

# .env 文件内容
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1

YOUR_HOLYSHEEP_API_KEY 替换成你在 HolySheep 注册后获得的具体密钥。注册地址在 这里,新用户会获得免费试用额度。

第一个工具调用程序:从 Hello World 开始

让我带你写一个最简单的工具调用程序。这个程序会让 AI 模型帮你执行一个简单的加法运算,同时展示工具调用的完整流程。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

加载环境变量

load_dotenv()

定义一个简单的计算工具

@tool def calculate(expression: str) -> str: """ 执行数学表达式计算 参数: expression: 要计算的数学表达式,如 "2 + 3" 或 "10 * 5" 返回: 计算结果的字符串形式 """ try: result = eval(expression) return f"计算结果:{expression} = {result}" except Exception as e: return f"计算错误:{str(e)}"

初始化模型(使用 HolySheep API)

llm = ChatOpenAI( model="gpt-4.1", base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"), api_key=os.getenv("HOLYSHEEP_API_KEY"), temperature=0.3 )

将工具绑定到模型

llm_with_tools = llm.bind_tools([calculate])

创建对话并触发工具调用

messages = [ {"role": "user", "content": "请帮我计算 (15 + 25) * 3 等于多少?"} ]

第一次调用:模型会识别需要使用计算工具

response = llm_with_tools.invoke(messages)

打印模型的响应(应该包含工具调用请求)

print("模型响应:") print(response) print("\n工具调用请求:") print(response.tool_calls)

运行这段代码后,你会看到模型返回了一个 tool_calls 列表,里面包含了要调用的工具名称和参数。这就是 LangChain 工具调用的核心机制——模型「决定」使用哪个工具,并给出具体的参数。

但在上面的代码中,工具还没有真正被执行。接下来我们要添加执行逻辑。

完整的工具调用流程:调用 → 执行 → 返回

下面是一个完整版的代码,包含了工具调用的全部流程:模型分析、工具执行、结果返回。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

load_dotenv()

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 这里用模拟数据,实际项目中可以调用真实天气 API
    weather_data = {
        "曼谷": "30°C,多云",
        "清迈": "28°C,晴天",
        "普吉": "31°C,有阵雨"
    }
    return weather_data.get(city, "未找到该城市的数据")

@tool
def get_time(timezone: str) -> str:
    """获取指定时区的当前时间"""
    from datetime import datetime
    try:
        import zoneinfo
        tz = zoneinfo.ZoneInfo(timezone)
        now = datetime.now(tz)
        return now.strftime("%Y年%m月%d日 %H:%M:%S")
    except:
        return "时区格式不正确,请使用如 'Asia/Bangkok' 的格式"

初始化模型

llm = ChatOpenAI( model="claude-sonnet-4.5", base_url="https://api.holysheep.ai/v1", api_key=os.getenv("HOLYSHEEP_API_KEY"), temperature=0.5 )

绑定工具

llm_with_tools = llm.bind_tools([get_weather, get_time])

完整的对话流程

def chat_with_tools(user_input: str): messages = [HumanMessage(content=user_input)] # 第一轮:模型决定是否调用工具 response = llm_with_tools.invoke(messages) messages.append(response) # 如果有工具调用,执行它们 if hasattr(response, 'tool_calls') and response.tool_calls: for tool_call in response.tool_calls: tool_name = tool_call['name'] tool_args = tool_call['args'] # 根据工具名称找到对应的函数 if tool_name == "get_weather": result = get_weather.invoke(tool_args) elif tool_name == "get_time": result = get_time.invoke(tool_args) else: result = "未知工具" # 将执行结果添加到消息历史 messages.append(ToolMessage(content=result, tool_call_id=tool_call['id'])) # 第二轮:模型根据工具结果生成最终回答 final_response = llm_with_tools.invoke(messages) return final_response.content return response.content

测试一下

print("=== 测试对话 ===\n") print("用户:请告诉我清迈现在的天气和时间") print(f"AI:{chat_with_tools('请告诉我清迈现在的天气和时间')}")

这个程序展示了工具调用的完整循环:用户提问 → 模型分析并决定调用工具 → 代码执行工具 → 结果返回给模型 → 模型生成自然语言回答。

我自己第一次跑通这个流程的时候特别兴奋,因为 AI 不只是「说话」了,而是真的能「做事」了。

MCP 协议进阶:自定义协议处理器

现在你已经掌握了基础用法,接下来看看如何用 MCP 协议扩展工具调用能力。MCP 的优势在于它支持复杂的工具定义、双向通信和流式响应。

import json
from typing import Any, Optional, List
from langchain_core.tools import BaseTool, StructuredTool
from pydantic import BaseModel, Field

定义 MCP 协议格式的工具输入输出模型

class MCPRequest(BaseModel): jsonrpc: str = "2.0" id: Optional[int] = None method: str params: Optional[dict] = {} class MCPResponse(BaseModel): jsonrpc: str = "2.0" id: Optional[int] = None result: Optional[Any] = None error: Optional[dict] = None

自定义 MCP 兼容的工具类

class MCPTool(StructuredTool): """支持 MCP 协议格式的工具基类""" def __init__(self, name: str, description: str, schema: type[BaseModel]): super().__init__( name=name, description=description, args_schema=schema ) def call_mcp(self, params: dict) -> MCPResponse: """以 MCP 协议格式调用工具""" try: result = self.invoke(params) return MCPResponse( jsonrpc="2.0", result=result, id=params.get("_request_id") ) except Exception as e: return MCPResponse( jsonrpc="2.0", error={ "code": -32603, "message": str(e) }, id=params.get("_request_id") )

创建 MCP 格式的汇率查询工具

class CurrencyInput(BaseModel): from_currency: str = Field(description="源货币代码,如 USD") to_currency: str = Field(description="目标货币代码,如 CNY") amount: float = Field(description="要转换的金额") currency_tool = MCPTool( name="currency_converter", description="将一种货币转换为另一种货币,支持所有主流货币", schema=CurrencyInput ) def currency_converter_func(from_currency: str, to_currency: str, amount: float) -> str: """货币转换的实际逻辑""" # 使用固定的模拟汇率(实际项目应调用真实 API) rates_to_usd = {"USD": 1.0, "CNY": 7.25, "THB": 35.5, "JPY": 155.0} if from_currency not in rates_to_usd or to_currency not in rates_to_usd: return f"不支持的货币代码:{from_currency} 或 {to_currency}" usd_amount = amount / rates_to_usd[from_currency] result = usd_amount * rates_to_usd[to_currency] return f"{amount} {from_currency} = {result:.2f} {to_currency}"

手动绑定工具函数

currency_tool._func = currency_converter_func

测试 MCP 格式调用

print("=== MCP 协议工具调用测试 ===\n")

模拟 MCP 请求

test_request = MCPRequest( id=1, method="tools/call", params={ "name": "currency_converter", "arguments": { "from_currency": "USD", "to_currency": "CNY", "amount": 100 }, "_request_id": 1 } ) print(f"MCP 请求:{test_request.model_dump_json(indent=2)}\n")

调用工具

response = currency_tool.call_mcp(test_request.params) print(f"MCP 响应:{response.model_dump_json(indent=2)}")

这段代码展示了如何构建符合 MCP 协议的扩展工具。通过自定义 MCPTool 类,我们可以让工具接受 MCP 标准格式的请求,并返回标准格式的响应。这在实际开发中非常有用,特别是当你要对接多个不同的 AI 模型或服务时。

多模型切换:用一个接口调用所有模型

HolySheep 的一个强大功能是支持多种主流模型。我来展示如何在代码中灵活切换不同的 AI 模型,全部使用同一个 HolySheep 接口。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

定义支持的模型配置

MODELS = { "gpt4.1": { "model": "gpt-4.1", "price_per_mtok": 8.00, # $8/MTok "strengths": ["代码生成", "复杂推理"] }, "claude-sonnet-4.5": { "model": "claude-sonnet-4.5", "price_per_mtok": 15.00, # $15/MTok "strengths": ["长文本理解", "创意写作"] }, "gemini-2.5-flash": { "model": "gemini-2.5-flash", "price_per_mtok": 2.50, # $2.50/MTok "strengths": ["快速响应", "多模态"] }, "deepseek-v3.2": { "model": "deepseek-v3.2", "price_per_mtok": 0.42, # $0.42/MTok "strengths": ["高性价比", "中文理解"] } } def create_model(model_key: str, **kwargs) -> ChatOpenAI: """创建指定模型的客户端""" if model_key not in MODELS: raise ValueError(f"未知模型:{model_key},可用模型:{list(MODELS.keys())}") config = MODELS[model_key] return ChatOpenAI( model=config["model"], base_url="https://api.holysheep.ai/v1", api_key=os.getenv("HOLYSHEEP_API_KEY"), **kwargs ) def ask_all_models(question: str): """向所有模型提问并比较回答""" results = {} for model_key, config in MODELS.items(): print(f"\n{'='*50}") print(f"模型:{config['model']}") print(f"价格:${config['price_per_mtok']}/MTok") print(f"优势:{', '.join(config['strengths'])}") print('-'*50) try: model = create_model(model_key, temperature=0.7) response = model.invoke(question) results[model_key] = response.content print(f"回答:{response.content[:200]}...") # 显示前200字 except Exception as e: print(f"错误:{str(e)}") results[model_key] = f"调用失败:{str(e)}" return results

测试:向所有模型询问同一个问题

print("=== 多模型对比测试 ===") question = "请用一句话解释什么是人工智能" results = ask_all_models(question) print("\n" + "="*60) print("总结:所有模型的回答已获取完成") print("推荐:根据回答质量选择合适的模型")

这段代码展示了 HolySheep 的一个重要优势——一个接口调用所有主流模型。根据实际测试,DeepSeek V3.2 的性价比最高($0.42/MTok),而 Gemini 2.5 Flash 的响应速度最快,延迟可以控制在 45 毫秒左右。

实战项目:构建一个智能助手

让我们用一个完整的项目来整合所有知识点。我要创建一个简单的智能助手,可以回答问题、查询信息、执行计算。

# smart_assistant.py

智能助手完整实现

import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage, SystemMessage load_dotenv()

========== 定义各种工具 ==========

@tool def calculator(expression: str) -> str: """计算数学表达式,支持 + - * / 和括号""" try: # 注意:eval 在生产环境中存在安全风险,这里仅用于演示 result = eval(expression) return f"✓ 计算完成:{expression} = {result}" except SyntaxError: return "✗ 表达式语法错误" except ZeroDivisionError: return "✗ 除数不能为零" except Exception as e: return f"✗ 计算错误:{str(e)}" @tool def search_knowledge(topic: str) -> str: """搜索知识库获取相关信息""" knowledge_base = { "python": "Python 是一种高级编程语言,以简洁易读的语法著称。", "langchain": "LangChain 是一个用于构建 AI 应用的框架,支持工具调用、链式调用等功能。", "mcp": "MCP(Model Context Protocol)是一种标准化协议,用于连接 AI 模型和各种外部工具。", "holysheep": "HolySheheep AI 是一个多模型 API 服务平台,支持 GPT、Claude、Gemini 等模型。" } topic_lower = topic.lower() if topic_lower in knowledge_base: return knowledge_base[topic_lower] return f"知识库中未找到关于「{topic}」的信息。" @tool def translate(text: str, target_lang: str = "English") -> str: """简单的翻译工具(模拟实现)""" translations = { ("你好", "English"): "Hello", ("谢谢", "English"): "Thank you", ("再见", "English"): "Goodbye", ("hello", "中文"): "你好", ("thank you", "中文"): "谢谢" } key = (text.lower(), target_lang.lower()) return translations.get(key, f"[模拟翻译] {text} → {target_lang}")

========== 初始化助手 ==========

class SmartAssistant: def __init__(self, model_name: str = "deepseek-v3.2"): self.llm = ChatOpenAI( model=model_name, base_url="https://api.holysheep.ai/v1", api_key=os.getenv("HOLYSHEEP_API_KEY"), temperature=0.7 ) # 绑定所有工具 self.tools = [calculator, search_knowledge, translate] self.llm_with_tools = self.llm.bind_tools(self.tools) # 系统提示词 self.system = SystemMessage(content="""你是一个智能助手,名为「小助」。 你可以帮助用户完成以下任务: 1. 计算数学表达式 2. 回答关于 Python、LangChain、MCP 等技术话题的问题 3. 进行简单的翻译 请始终保持友好和专业的态度。""") def chat(self, user_input: str) -> str: """处理用户输入并返回回答""" messages = [self.system, HumanMessage(content=user_input)] # 第一次调用:模型决定是否使用工具 response = self.llm_with_tools.invoke(messages) # 如果有工具调用请求,执行它们 if hasattr(response, 'tool_calls') and response.tool_calls: messages.append(response) for tool_call in response.tool_calls: tool_name = tool_call['name'] tool_args = tool_call['args'] # 查找并执行对应工具 for tool in self.tools: if tool.name == tool_name: result = tool.invoke(tool_args) break else: result = f"未知工具:{tool_name}" messages.append(ToolMessage( content=result, tool_call_id=tool_call['id'] )) # 第二次调用:模型根据工具结果生成回答 final_response = self.llm_with_tools.invoke(messages) return final_response.content return response.content

========== 运行助手 ==========

if __name__ == "__main__": assistant = SmartAssistant(model_name="deepseek-v3.2") # 测试各种功能 test_questions = [ "你好,请介绍一下你自己", "请计算 (100 + 200) * 2 / 4", "告诉我关于 LangChain 的知识", "请把「你好」翻译成英文" ] print("🤖 智能助手启动中...\n") for i, question in enumerate(test_questions, 1): print(f"【问题 {i}】{question}") print(f"【回答】{assistant.chat(question)}") print("-" * 60)

把这个文件保存为 smart_assistant.py,然后运行 python smart_assistant.py,你就能看到一个完整的多功能 AI 助手了。从我的测试经验来看,DeepSeek V3.2 模型在这个场景下表现很好,响应速度快而且回答准确。

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

在开发和调试过程中,我遇到了很多常见的错误。下面总结三个最常碰到的案例及其解决方案。

กรณีที่ 1: AuthenticationError - API Key ไม่ถูกต้อง

อาการ: 程序运行时出现 AuthenticationError401 Unauthorized 错误。

สาเหตุ: API 密钥配置错误或未正确加载。

วิธีแก้ไข:

# 错误写法:直接硬编码密钥(不安全)
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key="sk-xxxx-xxx-xxx"  # 这样写容易被误提交到 Git
)

正确写法:使用环境变量

import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件 llm = ChatOpenAI( model="gpt-4.1", base_url=os.environ.get("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"), api_key=os.environ.get("HOLYSHEEP_API_KEY") )

或者使用 pydantic-settings(更推荐)

pip install pydantic-settings

from pydantic_settings import BaseSettings class Settings(BaseSettings): holysheep_api_key: str holysheep_base_url: str = "https://api.holysheep.ai/v1" class Config: env_file = ".env" settings = Settings() print(f"API Key 已加载:{settings.holysheep_api_key[:8]}...") # 只显示前8位

กรณีที่ 2: ToolCallNotFound - 工具调用失败

อาการ: 模型返回了工具调用请求,但程序报错找不到工具。

สาเหตุ: 工具绑定时遗漏了某些工具,或者工具名称不匹配。

วิธีแก้ไข:

# 错误写法:工具列表和执行逻辑分离,容易遗漏
tools = [calculate, get_weather]  # 这里忘了加 translate

...后来在执行时...

for tool_call in response.tool_calls: tool_name = tool_call['name'] if tool_name == "calculate": result = calculate.invoke(tool_args) # 如果模型调用了 translate,这里就会报错!

正确写法:使用字典映射,确保所有工具都被处理

from functools import partial

定义所有工具的映射表

TOOL_REGISTRY = { "calculate": calculate, "get_weather": get_weather, "translate": translate, "search_knowledge": search_knowledge } def execute_tool_call(tool_call: dict): """统一执行工具调用的函数""" tool_name = tool_call['name'] tool_args = tool_call['args'] if tool_name not in TOOL_REGISTRY: return f"错误:未知工具 '{tool_name}',可用工具:{list(TOOL_REGISTRY.keys())}" try: tool = TOOL_REGISTRY[tool_name] result = tool.invoke(tool_args) return result except Exception as e: return f"工具执行错误:{str(e)}"

使用

for tool_call in response.tool_calls: result = execute_tool_call(tool_call) print(f"工具 {tool_call['name']} 执行结果:{result}")

กรณีที่ 3: RateLimitError - 请求频率超限

อาการ: 频繁调用时出现 RateLimitError429 Too Many Requests

สาเหตุ: 请求频率超过了 API 的限制。

วิธีแก้ไข:

import time
from tenacity import retry, stop_after_attempt, wait_exponential

方法一:使用 tenacity 库实现自动重试

@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) def call_with_retry(llm, messages): """带重试机制的 API 调用""" return llm.invoke(messages)

方法二:手动实现重试逻辑

def call_with_backoff(llm, messages, max_retries=3): """指数退避重试""" for attempt in range(max_retries): try: response = llm.invoke(messages) return response except Exception as e: if "rate_limit" in str(e).lower() and attempt < max_retries - 1: wait_time = 2 ** attempt # 2, 4, 8 秒 print(f"触发频率限制,等待 {wait_time} 秒后重试...") time.sleep(wait_time) else: raise e

方法三:添加请求间隔(最简单)

def call_with_delay(llm, messages, delay=0.5): """在请求之间添加延迟""" time.sleep(delay) return llm.invoke(messages)

使用示例

print("开始调用 API(带速率限制处理)...") response = call_with_backoff(llm_with_tools, messages) print("调用成功!")

总结与下一步建议

通过这篇教程,你应该已经掌握了 LangChain 工具调用的核心概念和实战技巧。我们从基础的环境配置开始,一步步实现了简单工具调用、完整调用流程、MCP 协议扩展、多模型切换,最后构建了一个可用的智能助手。

关键要点回顾:

如果你想继续深入学习,我建议尝试以下方向:

实践是最好的学习方式。建议你从修改本文的代码开始,添加自己的想法和功能。遇到问题时,可以查看 LangChain 官方文档或 HolySheheep 的 API 说明。

如果你觉得这篇文章有帮助,欢迎分享给其他想要学习 AI 应用开发的朋友!

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน