4.1 工具调用三部曲¶
学习目标¶
- 了解直接提示词做工具调用的方法和缺点
- 掌握使用Function Calling做工具调用的方法和缺点
- 掌握使用MCP协议做工具调用的方法和缺点
一、工具调用的原理¶
在所有的组件中，工具调用是比较关键的一环，它的整个流程可以表示为下图：
其主要工作步骤：
第①步：Agent 程序是我们开发的 AI 程序，在程序中会预先向大模型注册外部函数接口（一般不超过 20 个）。
当然，不同的大模型实现方式有所区别，例如DeepSeek是在发起请求（第②步）时直接将可能用到的函数列表直接发送给大模型。
第②步：用户使用自然语言（Prompt）发起请求，Agent 接收到请求。
第③步：Agent 程序将用户请求（Prompt）提交给大模型，大模型解析语义并根据第①步注册的函数信息，评估是否需要调用外部函数（Functions）。
第④步：模型如果判断需要调用函数，则生成包含函数 ID 和输入参数的调用指令，并返回给 Agent 程序。
第⑤步：Agent 程序接收到模型返回的调用指令后，执行对工具函数的调用。
第⑥步：工具函数执行后将结果返回给 Agent 程序。
第⑦步：Agent 程序将函数返回的结果和自定义提示词一起作为Prompt发送给大模型。
第⑧步：大模型结合函数返回的数据与上一轮上下文，生成最终结果，并返回给 Agent 。
第⑨步：Agent 程序将结果输出呈现给终端用户。
二、直接使用大模型和提示词¶
1 实例演示¶
参见第一章的调用工具
2 遗留问题¶
这种调用是基于提示词的，效果不稳定，难以投入工业化生产
- 不同人写提示词的水平不一致
- 大模型没有经过工具调用方面的专业训练，有较高的概率出错
急需解决调用提示词的规范性问题
三、使用Function Calling¶
写提示词特别强调输入输出的规范化，能否提供一种通用格式，通过严格限制大模型调用工具时的输入输出，来保证高度的一致性。
1 实例演示¶
import ollama
def callModel(prompt):
response = ollama.chat(
model='qwen2.5:7b',
messages=[{'role': 'user', 'content': prompt}],
tools=[
{
"type": "function",
"function": {
"name": "getTrainSchedule",
"description": "根据指定的日期、起点城市名称和终点城市名称，查询列车班次",
"parameters": {
"type": "object",
"properties": {
"queryDate": {
"type": "string",
"description": "指定日期",
},
"start": {
"type": "string",
"description": "起点城市名称",
},
"end": {
"type": "string",
"description": "终点城市名称",
},
},
"required": ["queryDate", "start", "end"],
},
}
}
]
)
#print(response['message']['content'])
#return response['message']['content']
return response
def checkTools(query):
toolCall = callModel(query)
return toolCall
def callTools(toolCall):
with open('tools.py', 'r', encoding='utf-8') as f:
content = f.read()
call = toolCall[0]
print(call.function)
print(call.function.arguments)
argsStr = ",".join(str(key)+'="'+call.function.arguments[key]+'"' for key in call.function.arguments)
print(argsStr)
content+="\ntoolRes="+str(call.function.name)+"("+argsStr+")"
print(content)
exec(content)
toolResult = locals()["toolRes"]
return toolResult
def getAnswer(query):
toolCall = []
response = checkTools(query)
if 'tool_calls' in response['message']:
toolCall = response['message']['tool_calls']
print("checkTools results:*********************************")
print(toolCall)
if len(toolCall)>0:
toolRes = callTools(toolCall)
print("callTools results:*********************************")
print(toolRes)
prompt = "你是一位可靠的个人助理，用户的问题是："+query+"。 查询工具得到的结果是："+toolRes+"。 根据这些信息，给出合理的回复。"
finalResult = callModel(prompt)
else:
prompt = "你是一位可靠的个人助理，用户的问题是：" + query + "。 根据这些信息，给出合理的回复。"
finalResult = callModel(prompt)
return finalResult['message']['content']
if __name__ == '__main__':
query = "我要在本周六到北京旅行，帮我规划一下列车班次，要求尽可能快捷舒适。"
result = getAnswer(query)
print(result)
2 遗留问题¶
-
在前面的所有工具调用例子中，工具描述都是由大模型应用的开发者来负责的。这里引入两个问题：
-
每个人都由自己对工具的理解，每个人的描述方式不一致，怎样保证工具调用的质量是稳定的？
- 每次开发新应用，都要重新写一遍工具描述，引入了太多重复的工作。
核心原因在于：对工具的描述是一个繁琐而复杂的过程，这项工作只有工具的开发者能够清晰准确地完成，而不应该由工具调用者负责。如果工具能清晰地描述自己的功能，就能够实现“一次编写，到处调用”，上述问题就迎刃而解了。
四、使用MCP协议¶
1 原理介绍¶
MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 推出的开源协议，旨在实现大型语言模型（LLM）与外部数据源和工具的无缝集成，用来在大模型和数据源之间建立安全双向的链接。有以下要点：
1.1 MCP 核心架构¶
MCP 遵循客户端-服务器架构（client-server），其中：
- MCP 主机（MCP Hosts）指的是发起请求的 LLM 应用程序。
- MCP 客户端（MCP Clients）指的是在主机程序内部，与 MCP server 保持 1:1 的连接。
- MCP 服务器（MCP Servers）指的是为 MCP client 提供上下文、工具和 prompt 信息。
- 本地资源（Local Resources）指的是本地计算机中可供 MCP server 安全访问的资源（例如文件、数据库）。
- 远程资源（Remote Resources）：MCP server 可以连接到的远程资源（例如通过 API）。
1.2 MCP Client¶
MCP client 充当 LLM 和 MCP server 之间的桥梁，MCP client 的工作流程如下：
- MCP client 首先从 MCP server 获取可用的工具列表。
- 将用户的查询连同工具描述通过 function calling 一起发送给 LLM。
- LLM 决定是否需要使用工具以及使用哪些工具。
- 如果需要使用工具，MCP client 会通过 MCP server 执行相应的工具调用。
- 工具调用的结果会被发送回 LLM。
- LLM 基于所有信息生成自然语言响应。
- 最后将响应展示给用户。
1.3 MCP Server¶
MCP server 是 MCP 架构中的关键组件，它可以提供 3 种主要类型的功能：
- 资源（Resources）：类似文件的数据，可以被客户端读取，如 API 响应或文件内容。
- 工具（Tools）：可以被 LLM 调用的函数（需要用户批准）。
- 提示（Prompts）：预先编写的模板，帮助用户完成特定任务。
这些功能使 MCP server 能够为 AI 应用提供丰富的上下文信息和操作能力，从而增强 LLM 的实用性和灵活性。
2.4 通信机制¶
MCP 协议支持两种主要的通信机制：基于标准输入输出的本地通信和基于SSE（Server-Sent Events）的远程通信。
这两种机制都使用 JSON-RPC 2.0格式进行消息传输，确保了通信的标准化和可扩展性。
- 本地通信：通过 stdio 传输数据，适用于在同一台机器上运行的客户端和服务器之间的通信。
- 远程通信：利用 SSE 与 HTTP 结合，实现跨网络的实时数据传输，适用于需要访问远程资源或分布式部署的场景。
2 实例演示¶
MCP server
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("test mcp")
@mcp.tool()
def getTrainSchedule(queryDate,start,end):
"""根据指定的日期、起点城市名称和终点城市名称，查询列车班次"""
print("获取列车时刻表")
result = [["D81","12:24","14:30","北京西站","540"],["K4427","15:38","21:30","北京站","220"]]
resultStr = "您查询的在 "+queryDate+" 这一天从 "+start+" 到 "+end+" 的列车共有 "+str(len(result))+"班：\n"
for res in result:
resultStr+=res[0]+" 次列车：发车时间为: "+res[1]+" 到站时间为： "+res[2]+" 发车站为： "+res[3]+" 票价为： "+res[4]+"\n"
return resultStr
if __name__ == "__main__":
mcp.run(transport='stdio')
MCP client
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 定义服务端参数：假设服务端脚本为 server.py
server_params = StdioServerParameters(
command=r"C:\Users\foxba\.conda\envs\langgraph2\python",
args=["mcpServer.py"],
env=None
)
# 使用 stdio_client 连接到服务端
async with stdio_client(server_params) as (stdio, write):
# 创建客户端会话
async with ClientSession(stdio, write) as session:
# 初始化会话
await session.initialize()
# 列出可用工具
tools_response = await session.list_tools()
print("可用工具:", [tool.name for tool in tools_response.tools])
available_tools = [{
"type":"function",
"function":{
"name":tool.name,
"description":tool.description,
"input_schema":tool.inputSchema
},
} for tool in tools_response.tools]
print(available_tools)
# 调用 getTrainSchedule 工具
result = await session.call_tool("getTrainSchedule", {"queryDate": "2025-10-15", "start": "青岛","end":"北京"})
print("工具调用结果:", result)
# 运行异步客户端
if __name__ == "__main__":
asyncio.run(main())