星月喜品工坊 LV
发表于 2025-4-8 13:31:40
无非就是开发function calling是更方面了一些，目前已经有很多mcp client的开源工具，例如：anthropic的Claude desktop 以及 cursor 、Chatwise 这些。自己只需要注册一个自己的工具。或者直接用其他人开源的mcp server。
先说结论：不会有什么实质性的影响，想要给行业带来颠覆性的影响，主要还是靠大语言模型的能力。哪有其他人回答的那么夸张。
让大模型调用的函数，分为两个过程：
- 大模型识别出需要调用的函数和参数
- 自己调用后把函数的结果和问题传递给大模型整理出答案
mcp就是方便你写注册函数和函数调用的过程。例如下面代码实现一个MCP server：- from mcp.server.fastmcp import FastMCP
- mcp = FastMCP("WeatherServer")
- @mcp.tool()
- async def query_weather(city: str) -> str:
- """
- 输入指定城市的英文名称，返回今日天气查询结果。
- :param city: 城市名称（需使用英文）
- :return: 格式化后的天气信息
- """
- data = await fetch_weather(city)
- return format_weather(data)
- async def process_query(self, query: str) -> str:
- messages = [{"role": "user", "content": query}]
-
- response = await self.session.list_tools()
-
- available_tools = [{
- "type": "function",
- "function": {
- "name": tool.name,
- "description": tool.description,
- "input_schema": tool.inputSchema
- }
- } for tool in response.tools]
-
- response = self.client.chat.completions.create(
- model=self.model,
- messages=messages,
- tools=available_tools
- )
- content = response.choices[0]
- if content.finish_reason == "tool_calls":
- # 解析工具
- tool_call = content.message.tool_calls[0]
- tool_name = tool_call.function.name
- tool_args = json.loads(tool_call.function.arguments)
-
- # 执行工具
- result = await self.session.call_tool(tool_name, tool_args)
- print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
-
- messages.append(content.message.model_dump())
- messages.append({
- "role": "tool",
- "content": result.content[0].text,
- "tool_call_id": tool_call.id,
- })
-
- response = self.client.chat.completions.create(
- model=self.model,
- messages=messages,
- )
- return response.choices[0].message.content
-
- return content.message.content
同样的，不是用mcp时，也可以用下面代码来简化注册工具- import json
- import inspect
- from datetime import datetime, timedelta
- from collections import OrderedDict
- from types import GenericAlias
- from typing import get_origin, Annotated, Dict, List, TypeVar
- tools = []
- _TOOL_HOOKS = OrderedDict()
- def register_tools(func: callable):
- tool_name = func.__name__
- tool_description = inspect.getdoc(func).strip()
- python_params = inspect.signature(func).parameters
- tool_params = []
- for name, param in python_params.items():
- annotation = param.annotation
- if annotation is inspect.Parameter.empty:
- raise TypeError(f"Parameter `{name}` missing type annotation")
- if get_origin(annotation) != Annotated:
- raise TypeError(f"Annotation type for `{name}` must be typing.Annotated")
- typ, (description, required) = annotation.__origin__, annotation.__metadata__
- typ: str = str(typ) if isinstance(typ, GenericAlias) else typ.__name__
- if not isinstance(description, str):
- raise TypeError(f"Description for `{name}` must be a string")
- if not isinstance(required, bool):
- raise TypeError(f"Required for `{name}` must be a bool")
- tool_params.append({
- "name": name,
- "description": description,
- "type": typ,
- "required": required
- })
- properties = OrderedDict()
- for param in tool_params:
- properties[param["name"]] = {
- "description": param["description"],
- "type": param["type"]
- }
- tool_def = {
- "type": "function",
- "function": {
- "name": tool_name,
- "description": tool_description,
- "parameters": {
- "type": "object",
- "properties": properties,
- },
- "required": [tool_param["name"] for tool_param in tool_params if tool_param["required"]]
- },
-
- }
- tools.append(tool_def)
- _TOOL_HOOKS[tool_name] = func
-
- return func
- @register_tools
- async def query_weather(
- city: Annotated[string, '输入指定城市的英文名称', True]
- ) -> str:
- """
- 输入指定城市的英文名称，返回今日天气查询结果。
- Args:
- city: 输入指定城市的英文名称
-
- return: 格式化后的天气信息
- """
- data = await fetch_weather(city)
- return format_weather(data)
所以说mcp不会给行业带来什么实质性的影响。 |
|
