在 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 ไม่ถูกต้อง
อาการ: 程序运行时出现 AuthenticationError 或 401 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 - 请求频率超限
อาการ: 频繁调用时出现 RateLimitError 或 429 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 协议扩展、多模型切换,最后构建了一个可用的智能助手。
关键要点回顾:
- 工具调用的本质是让 AI 模型「决定」何时调用工具,然后由代码执行工具并返回结果
- MCP 协议提供了标准化的工具定义和调用格式,让跨模型开发变得简单
- 使用 HolySheep 可以用统一的接口访问多种模型,价格透明且性价比高
- 在实际开发中要注意错误处理、速率限制和安全性问题
如果你想继续深入学习,我建议尝试以下方向:
- 添加流式输出(Streaming)提升用户体验
- 实现对话历史管理,支持多轮对话
- 连接真实的外部 API(如天气、股票、新闻等)
- 学习 LangChain Expression Language (LCEL) 编写更复杂的 Chain
实践是最好的学习方式。建议你从修改本文的代码开始,添加自己的想法和功能。遇到问题时,可以查看 LangChain 官方文档或 HolySheheep 的 API 说明。
如果你觉得这篇文章有帮助,欢迎分享给其他想要学习 AI 应用开发的朋友!
👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน