admin 管理员组

文章数量: 1184232

MCP与Function Calling

视频链接：

MCP与Function Calling

一、MCP与Function Calling的关系误区及纠正

  • 普遍错误观点：认为MCP统一了Function Calling的协议，会取代Function Calling。
  • 正确关系：两者是互补关系，在大模型领域所处地位不同，会长期共存，甚至可出现在同一链路中。
  • 核心区别：MCP作用于服务器与函数之间，规定工具的发现与调用协议；Function Calling作用于模型（或模型API）与函数列表之间，是模型挑选函数的能力。

二、Function Calling的概念

  • 本质定义：指模型与外部工具交互的能力，外部工具本质是编程语言中的函数，因此Function Calling即模型调用函数的能力。
  • 关键特点：模型自身无法直接调用工具，需通过中间人（如服务器）协助完成调用。
  • 扩展理解：在实际应用中，更多关注模型API的Function Calling能力，因其与模型的Function Calling能力本质一致（均为从函数列表中挑选函数），只是视角不同。

三、Function Calling链路分析（以ChatGPT为例）

  1. 角色构成：用户、ChatGPT应用（含网络搜索函数、OpenAI服务器、GPT4O模型）。
  2. 流程步骤：
    • 用户发送问题（如“纽约明天的天气怎么样”）给ChatGPT应用。
    • OpenAI服务器将问题及可用工具（如网络搜索）转发给GPT4O模型。
    • 模型评估后决定调用工具，返回工具使用请求（含工具名称和参数）给服务器。
    • 服务器调用对应函数（如网络搜索），将结果返回给模型。
    • 模型根据结果总结答案，经服务器反馈给用户。
  3. 作用环节：聚焦于模型（或模型API）挑选工具、解析工具执行结果的环节，与函数的实际调用过程无关。

四、Function Calling协议内容分析

  • 协议核心：规定工具列表传给模型API的方式，以及模型API返回挑选的工具和参数的方式。
  • 通过Mark chat应用演示：
    • 应用流程：用户提问后，先调用search工具获取天气信息，再由模型总结答案。
    • 核心代码：process_user_query函数包含Function Calling的核心逻辑，涉及调用模型API、提取工具信息、执行工具、再次调用模型API等步骤。
    • 模型API请求字段：
      • model：指定使用的模型（如GPT4O）。
      • messages：包含用户问题等历史消息。
      • tools：可用工具列表，含工具名称及参数的JSON schema描述。
      • stream：是否采用流式返回（示例中设为false）。
    • 模型API返回内容：首次返回需调用的工具及参数，二次返回基于工具结果的最终答案。
      详细代码：GitHub 代码链接

class LLMProcessor:
    def __init__(self):
        self.api_key = OPENROUTER_API_KEY
        self.base_url = "https://openrouter.ai/api/v1/chat/completions"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        self.history = []

    def process_user_query(self, query):

        self.history.append({"role": "user", "content": query})

        first_model_response = self.call_model()

        first_model_message = first_model_response["choices"][0]["message"]
        self.history.append(first_model_message)

        # 检查模型是否需要调用工具
        if "tool_calls" in first_model_message and first_model_message["tool_calls"]:
            tool_call = first_model_message["tool_calls"][0]
            tool_name = tool_call["function"]["name"]
            tool_args = json.loads(tool_call["function"]["arguments"])

            result = self.execute_tool(tool_name, tool_args)

            self.history.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "name": tool_name,
                "content": result
            })

            second_response_data = self.call_model_after_tool_execution()

            final_message = second_response_data["choices"][0]["message"]
            self.history.append(final_message)

            return {
                "tool_name": tool_name,
                "tool_parameters": tool_args,
                "tool_executed": True,
                "tool_result": result,
                "final_response": final_message["content"],
            }
        else:
            return {
                "final_response": first_model_message["content"],
            }

    def execute_tool(self, function_name, args):
        if function_name == "search":
            # 正常情况下，这里应该调用相关 API 做搜索，为了减少代码的复杂度，
            # 这里我们返回一段假的工具执行结果，用以测试
            return "纽约市今天的天气是晴天，明天的天气是多云。"
        else:
            raise ValueError(f"未知的工具名称：{function_name}")

    def call_model(self):

        request_body = {
            "model": MODEL_NAME,
            "messages": self.history,
            "tools": TOOLS,
            "stream": False,
        }

        response = requests.post(
            self.base_url,
            headers=self.headers,
            json=request_body
        )

        logger.log(f"第一次模型请求：\n{json.dumps(request_body, indent=2, ensure_ascii=False)}\n")
        logger.log(f"第一次模型返回：\n{json.dumps(response.json(), indent=2, ensure_ascii=False)}\n")

        if response.status_code != 200:
            raise Exception(f"API request failed with status {response.status_code}: {response.text}")

        return response.json()

    def call_model_after_tool_execution(self):
        second_request_body = {
            "model": MODEL_NAME,
            "messages": self.history,
            "tools": TOOLS,
        }

        # Make the second POST request
        second_response = requests.post(
            self.base_url,
            headers=self.headers,
            json=second_request_body
        )

        logger.log(f"第二次模型请求：\n{json.dumps(second_request_body, indent=2, ensure_ascii=False)}\n")
        logger.log(f"第二次模型返回：\n{json.dumps(second_response.json(), indent=2, ensure_ascii=False)}\n")

        # Check if the request was successful
        if second_response.status_code != 200:
            raise Exception(f"API request failed with status {second_response.status_code}: {second_response.text}")

        # Parse the second response
        return second_response.json()

    def execute_tool_with_mcp(self, function_name, args):
        loop = asyncio.new_event_loop()
        return loop.run_until_complete(self.execute_tool_with_mcp_async(function_name, args))


    async def execute_tool_with_mcp_async(self, function_name, args):
        # 获取与当前脚本同目录下的 mcp_server.py 的绝对地址
        mcp_server_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "mcp_server.py"))

        # 启动 MCP Client 并调用 MCP Tool
        async with MCPClient("uv", ["run", mcp_server_path]) as client:
            return await client.call_tool(function_name, args)

模型交互日志：

第一次模型请求：
{
  "model": "openai/gpt-4o-mini",
  "messages": [
    {
      "role": "user",
      "content": "纽约明天天气怎么样"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "search",
        "description": "搜索网络",
        "parameters": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string",
              "description": "要搜索的内容"
            }
          },
          "required": [
            "query"
          ]
        }
      }
    }
  ],
  "stream": false
}

第一次模型返回：
{
  "id": "gen-1754493608-WYcfIsAolMwQUzxx5cYt",
  "provider": "OpenAI",
  "model": "openai/gpt-4o-mini",
  "object": "chatpletion",
  "created": 1754493608,
  "choices": [
    {
      "logprobs": null,
      "finish_reason": "tool_calls",
      "native_finish_reason": "tool_calls",
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "",
        "refusal": null,
        "reasoning": null,
        "tool_calls": [
          {
            "index": 0,
            "id": "call_pv9XkdO5Zirauhha3xk0JcMH",
            "type": "function",
            "function": {
              "name": "search",
              "arguments": "{\"query\":\"纽约 明天 天气预报\"}"
            }
          }
        ]
      }
    }
  ],
  "system_fingerprint": "fp_34a54ae93c",
  "usage": {
    "prompt_tokens": 51,
    "completion_tokens": 19,
    "total_tokens": 70,
    "prompt_tokens_details": {
      "cached_tokens": 0
    },
    "completion_tokens_details": {
      "reasoning_tokens": 0
    }
  }
}

第二次模型请求：
{
  "model": "openai/gpt-4o-mini",
  "messages": [
    {
      "role": "user",
      "content": "纽约明天天气怎么样"
    },
    {
      "role": "assistant",
      "content": "",
      "refusal": null,
      "reasoning": null,
      "tool_calls": [
        {
          "index": 0,
          "id": "call_pv9XkdO5Zirauhha3xk0JcMH",
          "type": "function",
          "function": {
            "name": "search",
            "arguments": "{\"query\":\"纽约 明天 天气预报\"}"
          }
        }
      ]
    },
    {
      "role": "tool",
      "tool_call_id": "call_pv9XkdO5Zirauhha3xk0JcMH",
      "name": "search",
      "content": "纽约市今天的天气是晴天，明天的天气是多云。"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "search",
        "description": "搜索网络",
        "parameters": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string",
              "description": "要搜索的内容"
            }
          },
          "required": [
            "query"
          ]
        }
      }
    }
  ]
}

第二次模型返回：
{
  "id": "gen-1754493610-nMcr5YRo7yeEAgJKwYZB",
  "provider": "OpenAI",
  "model": "openai/gpt-4o-mini",
  "object": "chatpletion",
  "created": 1754493610,
  "choices": [
    {
      "logprobs": null,
      "finish_reason": "stop",
      "native_finish_reason": "stop",
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "纽约明天的天气预报是多云。",
        "refusal": null,
        "reasoning": null
      }
    }
  ],
  "system_fingerprint": "fp_34a54ae93c",
  "usage": {
    "prompt_tokens": 94,
    "completion_tokens": 12,
    "total_tokens": 106,
    "prompt_tokens_details": {
      "cached_tokens": 0
    },
    "completion_tokens_details": {
      "reasoning_tokens": 0
    }
  }
}

FUCTION CALLING具体要点：就是是否具有解析和使用的能力。如果能力差，不一定能正确解析和返回正确的参数

五、同时使用Function Calling和MCP的实现

  • 修改要点：在Mark chat应用中，将执行工具的函数execute_tool替换为遵循MCP规范的execute_tool_with_mcp。
  • 实现逻辑：
    • execute_tool_with_mcp启动MCP客户端，连接项目中的MCP服务器。
    • MCP服务器返回带标识的结果，证明链路中已集成MCP。
  • 效果验证：修改后应用可正常返回结果，且工具调用信息显示来自MCP服务器，验证了两者可协同工作。

就是是需要改下面代码：

        if "tool_calls" in first_model_message and first_model_message["tool_calls"]:
            tool_call = first_model_message["tool_calls"][0]
            tool_name = tool_call["function"]["name"]
            tool_args = json.loads(tool_call["function"]["arguments"])

            result = self.execute_tool_with_mcp(tool_name, tool_args)

            self.history.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "name": tool_name,
                "content": result
            })

这调用MCP服务端代码如下（上面代码也有这里列出）：

def execute_tool_with_mcp(self, function_name, args):
    # 创建一个新的异步事件循环
    loop = asyncio.new_event_loop()
    # 运行异步函数execute_tool_with_mcp_async直到完成，并返回其结果
    # 这是同步函数调用异步函数的常见方式，通过事件循环驱动异步执行
    return loop.run_until_complete(self.execute_tool_with_mcp_async(function_name, args))


async def execute_tool_with_mcp_async(self, function_name, args):
    # 计算mcp_server.py的绝对路径：
    # 1. os.path.dirname(__file__) 获取当前脚本所在目录
    # 2. os.path.join(...) 拼接当前目录与"mcp_server.py"得到相对路径
    # 3. os.path.abspath(...) 将相对路径转换为绝对路径，确保跨环境可正确访问
    mcp_server_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "mcp_server.py"))

    # 使用异步上下文管理器启动MCP客户端：
    # 1. "uv" 是启动器（可能是uvicorn的简写，用于运行ASGI应用）
    # 2. ["run", mcp_server_path] 是传递给uv的参数，意为运行mcp_server.py
    # 3. async with ... 确保客户端使用完后自动关闭资源（如连接）
    async with MCPClient("uv", ["run", mcp_server_path]) as client:
        # 调用MCP客户端的call_tool方法，传入工具名和参数，等待结果返回
        # 这是实际执行工具调用的核心操作
        return await client.call_tool(function_name, args)

MCPSERVER 代码如下：和在进阶篇讲的运行MCPSERVER,并和cline进行交互一致

from mcp.server.fastmcp import FastMCP


# Initialize FastMCP server
mcp = FastMCP("search_mcp_server", log_level="ERROR")


# Constants
@mcp.tool()
async def search(query: str) -> str:
    """搜索网络

    Args:
        query: 搜索内容
    """
    # 正常情况下，这里应该调用相关 API 做搜索，为了减少代码的复杂度，
    # 这里我们返回一段假的工具执行结果，用以测试
    return "来自 MCP Server 的答案：纽约市今天的天气是晴天，明天的天气是多云。"


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')

本文标签： MCP function Calling

版权声明：本文标题：MCP与Function Calling 内容由网友自发贡献，该文观点仅代表作者本人， 转载请联系作者并注明出处：http://www.roclinux.cn/b/1765998180a3430767.html， 本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容，一经查实，本站将立刻删除。