大模型应用与扩展-Function Calling与MCP

大模型的交互过程

这是一个简单的提问-响应过程。为了解决模型的“幻觉问题”，可以通过多种方式向模型添加有用的信息来增加模型响应结果的可信度

1 . 多轮对话，通过多次提问过程，更加精确的描述用户的实际需求。如Agent

2 . 客户端代理对用户请求进行扩展，最终请求 携带 代理描述信息+原用户请求，向大模型发起请求。如RAG，MCP

3 . 模型端 功能扩展，模型端使用本地工具（一般是一些方法调用）来自行获取扩展信息，增强相应质量。如Function Calling

综上，在从流程上，大模型的扩展会有本地端的功能增强，服务端功能增强两种方式。从内容上，则分为 调用 / 内容 两种。

Function与MCP

我们知道 ，只有每一次请求的发起都会被正确转化为对特定功能的调用，这时AI 所能提供的服务才可以作为稳定系统的一部分。在这个角度，大模型能否根据用户请求，进行稳定的参数调用响应。成为大模型能否参与到业务流程中的重要要求。

Function Calling

Function Calling 是 openai提供的一种序列化请求/响应的方式。通过向大模型发起结构化的调用请求，模型会分析请求内容，并响应对应的函数以及参数调用。

MCP

MCP(Model Context Protocol) 模型上下文协议。它是一种与模型的通信方式。其定义了MCP Client 和MCP Sever 两种角色。也可以将MCP理解为一个位于本地的中间过程代理。通过接收用户向模型发起的请求，之后查询本地支持的API字段，组装后向大模型发起请求，大模型接收后分析用户请求以及其中定义的tools信息。响应一个方法调用请求（function call 在这个步骤被执行）给MCP server. MCP server 接收后。根据调用请求在本地执行对应的方法，获取返回信息，再传递给大模型处理。大模型处理后，整理结果向用户返回。

MCP 实现

客户端

MCP 架构 定义了 3个过程角色 ：

1. MCP HOST : 主要负责与用户进行交互。接收用户发出的消息请求
2. MCP Client : 组织message，构造发送给模型的请求。一般将MCP HOST 包含在客户端内
3. MCP Server : 维护本地的方法调用列表，处理接收到的模型功能调用，将处理后的消息和用户消息再一次发送给大模型

一个标准的大模型HTTP请求（基于qwen3）类似于：

{
    "model": "qwen3-8b",  //  指定模型名称
    "input": {     // 传入的消息
        "messages": [   // 传入的消息内容
            {
                "role": "system",   // 定义消息的角色，系统prompt
                "content": [   
                    {
                        "type": "text",  // 定义消息类型
                        "text": "你是一个助手，你需要根据用户的输入进行回答。用户的输入中包含它所能提供的调用功能，如果你觉得执行某个功能，会帮助得到更 好的结果，请向用户返回对应的json请求"  // 具体的文本信息
                    }
                ]
            },
                "role": "user",      // 消息角色，一般是用户请求
                "content": [
                    {
                        "type": "text",   // 消息类型 
                        "text": "我想执行test"  // 用户角色
                    }
                ]
            }
        ]
    },
    "parameters": {
        "temperature": 0.7,   // 温度，->0 表示生成的内容更发散， ->1 表示生成的内容更稳定
        "top_p": 0.8,  // 表示相关度 
        "result_format": "message",  // 响应的信息格式
        "enable_thinking": False  // qwen特有参数
    },
    "tools": [  // 本地的工具声明
        {
            "type": "function",   // 类型指定为function,声明为一个可执行函数的信息
            "function": {           
                "name": "test",   // 函数名
                "description": "调用test功能", // 对函数功能的描述
                "parameters": {  // 参数描述
                    "type": "object",  
                    "properties": {
                        "name": {
                            "type": "string",
                            "description": "name参数"
                        }
                    },
                    "required": [
                        "name"  // 要求的参数信息
                    ]
                }
            }
        }
    ],
    "tool_choice": "auto"
}

qwen3 支持的 role 角色定义 （经过测试目前 tool_call的Function Calling 机制仍未能正确识别）

role 值	含义说明	示例场景
system	设置系统行为和角色，引导模型回答风格或身份	“你是一个有耐心的英语老师”
user	用户输入，用于发起请求或提问	“请告诉我北京今天的天气”
assistant	模型生成的回答内容	自动由模型生成返回消息
tool_call	（可选）表示模型请求调用某个工具（函数）	用于 Function Calling 机制
tool_response	（可选）表示工具调用的响应结果，传给模型继续生成	工具调用返回值的嵌入，例如天气数据

首先定义了一个MCPmessage的 结构，在后续的传递过程中通过创建MCPmessage对象来完成用户，响应消息的插入

class MCPmessage:
    def __init__(self,role,type,message):
        self.data = {
            "role":role,
            "content":[{
                "type":type,
                type:message
            }]
        }

客户端实现

class LLMClient:
    """大模型客户端"""
  
    def __init__(self):
        self.api_key = "your-api-key"
        self.base_url = "<https://dashscope.aliyuncs.com/api/v1>"
        self.model = "qwen-8b"
  
    def configure(self, api_key: str, base_url: str = None, model: str = None):
        """配置大模型参数"""
        self.api_key = api_key
        if base_url:
            self.base_url = base_url
        if model:
            self.model = model
  
    async def chat_completion(self, messages, tools) :
        """调用大模型进行对话"""
        if not self.api_key:
            return {"error": "API Key未配置"}
        messages.insert(0,{
                        "role": "system",
                        "content": [
                            {
                                "type": "text",
                                "text": "你是一个助手，你需要根据用户的输入进行回答。用户的输入中包含它所能提供的调用功能，如果你觉得执行某个功能，会帮助得到更好的结果，请向用户返回对应的json请求,你可以调用如下功能，如果发现“执行test”等关键词，请调用 execute_test，并传入 name=\\"test\\"",
                            }
                        ],
                    })

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
      
        payload = {
            "model": self.model,
            "input" : {
                "messages": messages
            },
            "parameters": {
                "temperature": 0.7,
                "top_p": 0.8,
                "result_format": "tool_call",
                "enable_thinking": False,
            },
            "tools": tools,
            "tool_choice": "auto"
        }
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"
        try:
            print(payload)
            response = requests.post(
                f"{self.base_url}/services/aigc/text-generation/generation",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": f"大模型调用失败: {str(e)}"}

MCP 服务器

MCP 服务器主要负责 本地的功能管理，以及中间过程方法调用的功能处理。

即当 用户发起的请求信息 得到模型响应了一个调用传递后，会根据传递的JSON请求，解析识别到对应的函数进行执行，并将结果与用户原请求一起返回给大模型由大模型继续处理。

从这里我们可以看出来，大模型本身并不保存会话记录。所谓的”会话“ 实际上是每一次请求都携带了之前请求的全部token内容，由大模型依据所有内容再进行生成。

class McpServer :
    def __init__(self):
        self.llmclient = LLMClient()
        self.conversation_history = []
        self.session_id = str(uuid.uuid4())
        self.api_registry = {}

    def convert_functions_to_tools(self):
        """将函数注册表转换为API工具格式"""
        tools = []
        for name, func_struct in self.api_registry.items():
            tool = {
                "type": "function",
                "function": {
                    "name": name,
                    "description": f"调用{name}功能",
                    "parameters": {
                        "type": "object",
                        "properties": {},
                        "required": func_struct.parameters_list
                    }
                }
            }
            # 为每个参数添加属性定义
            for param in func_struct.parameters_list:
                tool["function"]["parameters"]["properties"][param] = {
                    "type": "string",
                    "description": f"{param}参数"
                }
            tools.append(tool)
        return tools

    async def process_message(self,user_message) :
		    """对请求预处理，并向大模型发送请求"""
        message = MCPmessage(role="user", type="text", message=user_message)
        self.conversation_history.append(message.data)
        self.messages = []
        for i in self.conversation_history[-3:]:
            self.messages.append(i) 
        tools=self.convert_functions_to_tools()
        response = await self.llmrequest(self.messages,tools)  

    def search_parameter(self, message):
		    """因为目前qwen3 对于function calling 的 tool_call 字段响应不够稳定，
		    通过一个解析方法来获取方法名和参数信息"""
        start, end = -1, -1
        for i, char in enumerate(message):  # 修复：正确遍历字符
            if char == "\\"":
                if start == -1:
                    start = i + 1  # 跳过引号本身
                else:
                    end = i
                    break
      
        if start != -1 and end != -1:
            return message[start:end]
        return ""
    async def llmrequest(self,messages,tools):
		    """用户请求->模型调用响应->本地方法执行->结果回传->模型响应处理流程"""
        response = await self.llmclient.chat_completion(messages,tools)
        # 获取用户的请求的响应信息
        try :
            FunctionCallMessage = response["output"]["choices"][0]["message"]["content"]
            # 检查响应的内容
            if "action" in FunctionCallMessage:
            # 当前无法稳定返回 "tool_call"。 先通过 检索实现 方法调用
                function_name = self.search_parameter(FunctionCallMessage.split("\\"action\\"")[1])
                # 使用检索函数获取待执行函数名
                if function_name in self.api_registry.keys():
                # 查询维护的方法列表中是否存在对应的方法
                    target_function = self.api_registry[function_name]
                    # 查询到对应的对象
                    parameters_list = []
                    for i  in target_function.parameters_list:
                        if i in FunctionCallMessage: 
                            parameters_list.append(self.search_parameter(FunctionCallMessage.split(i)[-1]))
						                # 从响应结果中提取方法
                self.api_registry[function_name].function(parameters_list)
                # 执行函数调用
                response = requests.post(self.api_registry[function_name].url)
                # 如果是一个接口请求，则可以使用requests向接口发出请求
                self.conversation_history.append(MCPmessage("function","text","test 功能已经执行完毕，请向客户端返回成功信息"  ).data)
                # 将响应结果添加到对话历史中
                response = await self.llmclient.chat_completion([self.conversation_history[-1]],tools)              
				        # 获取响应
        except KeyError:
            pass 
        return response

其他

定义方法的结构维护，便于查询方法和方法的参数信息。

class FunctionStruct :
    def __init__(self,name,function,parameters_list,url):
        self.name = name
        self.function = function
        self.parameters_list = parameters_list
        self.url = url

def execute_test(*args):
    print("function test")
    return None

def execute(*args):
    print("function execute")
    return None

def main():
    ms = McpServer()
    ms.api_registry["execute_test"] = FunctionStruct("execute_test", execute_test, ["name"], "<http://www.baidu.com>")
    ms.api_registry["execute"] = FunctionStruct("execute", execute, ["name"], "<http://www.baidu.com>")
  
    async def async_interaction():
        while True:
            print("请输入你的问题：")
            question = input()
            await ms.process_message(question)
  
    asyncio.run(async_interaction())
if __name__ == "__main__":
    main()

结果

通过打印接口信息可以看到对大模型发起了两次请求，并最终执行了功能，并返回成功的响应

请输入你的问题：
我想执行test
{'model': 'qwen3-8b', 'input': {'messages': [{'role': 'system', 'content': [{'type': 'text', 'text': '你是一个助手，你需要根据用户的输入进行回答。用户的输入中包含它所能提供的调用功能，如果你觉得执行某个功能，会帮助得到更好的结果，请向用 户返回对应的json请求,你可以调用如下功能，如果发现“执行test”等关键词，请调用 execute_test，并传入 name="test"'}]}, {'role': 'user', 'content': [{'type': 'text', 'text': '我想执行test'}]}, {'role': 'user', 'content': [{'type': 'text', 'text': '我想执行test'}]}, {'role': 'user', 'content': [{'type': 'text', 'text': '我想执行test'}]}]}, 'parameters': {'temperature': 0.7, 'top_p': 0.8, 'result_format': 'message', 'enable_thinking': False}, 'tools': [{'type': 'function', 'function': {'name': 'execute_test', 'description': '调用execute_test功能', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string', 'description': 'name参数'}}, 'required': ['name']}}}, {'type': 'function', 'function': {'name': 'execute', 'description': '调用execute功能', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string', 'description': 'name参数'}}, 'required': ['name']}}}], 'tool_choice': 'auto'}
{'output': {'choices': [{'finish_reason': 'stop', 'message': {'role': 'assistant', 'content': '{"action": "execute_test", "name": "test"}'}}]}, 'usage': {'total_tokens': 108, 'output_tokens': 13, 'input_tokens': 95}, 'request_id': '09dc09ea-ed80-9632-bd1d-77d05cde942e'}
function test
{'model': 'qwen3-8b', 'input': {'messages': [{'role': 'system', 'content': [{'type': 'text', 'text': '你是一个助手，你需要根据用户的输入进行回答。用户的输入中包含它所能提供的调用功能，如果你觉得执行某个功能，会帮助得到更好的结果，请向用 户返回对应的json请求,你可以调用如下功能，如果发现“执行test”等关键词，请调用 execute_test，并传入 name="test"'}]}, {'role': 'function', 'content': [{'type': 'text', 'text': 'test 功能已经执行完毕，请向客户端返回成功信息'}]}]}, 'parameters': {'temperature': 0.7, 'top_p': 0.8, 'result_format': 'message', 'enable_thinking': False}, 'tools': [{'type': 'function', 'function': {'name': 'execute_test', 'description': '调用execute_test功能', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string', 'description': 'name参数'}}, 'required': ['name']}}}, {'type': 'function', 'function': {'name': 'execute', 'description': '调用execute功能', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string', 'description': 'name参数'}}, 'required': ['name']}}}], 'tool_choice': 'auto'}
```json
{
  "status": "success",
  "message": "test 功能已经执行完毕"
}

方法调用链

总结

基于以上尝试，我们可以看到MCP 进行了一个中间处理过程，通过LLM 来进行功能与数组的组织最终返回给用户自然语言的结果描述。而不必再关注具体的中间过程。

与 RAG 技术相比，MCP 提供了另一种大模型能力扩展路径。RAG（Retrieval-Augmented Generation）本质上是一种 静态的、基于预先构建的向量数据库进行信息增强的机制 ，它能提供相对 稳定且相关性明确的知识补充 ，适用于对资料完整性要求较高的场景。但其能力受限于数据库的更新频率和覆盖范围。

而 MCP 更侧重于 动态信息的实时获取与响应处理 ，通过将本地方法封装为函数工具，模型可以根据语义分析结果主动发起实时调用，从而打破传统问答的静态信息局限。这种机制将对话式交互转化为 任务式的数据动态组织流程 ，具备更强的灵活性和时效性，尤其适用于实时数据接口、系统调用和复杂业务流程的交互式处理场景。

因此，两者并非冲突关系，而是互为补充 —— RAG 提供 知识的可控性与一致性 ，MCP 提供 执行的实时性与系统集成能力 。在实际系统中可根据场景组合使用，构建更智能、更可靠的 AI 代理系统。