Kimi K2 工具调用（Tool Calling）开发指南：从 OpenAI 兼容接口到流式与手动解析实战-育师

Kimi K2 工具调用（Tool Calling）开发指南：从 OpenAI 兼容接口到流式与手动解析实战

【免费下载链接】Kimi-K2Kimi K2 is the large language model series developed by Moonshot AI team项目地址: https://gitcode.com/GitHub_Trending/ki/Kimi-K2

Kimi K2（moonshotai/Kimi-K2）是 Moonshot AI 团队开源的大规模 MoE 语言模型，其 Instruct 版本专为 agentic 场景优化，具备较强的工具调用能力。本文以仓库 docs/tool_call_guidance.md 为核心，完整讲解 Kimi K2 工具调用的端到端流程：如何向模型声明函数、如何在非流式/流式对话中驱动模型自主调用工具、如何在没有解析器的框架中手动解析调用请求，并结合 docs/deploy_guidance.md 说明部署侧必须开启的解析器选项。读完本文，你将能够基于 OpenAI 兼容接口或原生 completions 接口，为 Kimi K2 接入任意自定义工具，构建可落地的 Agent 应用。

工具调用前置条件：部署侧必须开启解析器

Kimi K2 的工具调用能力依赖推理服务端的"工具调用解析器"（tool call parser）。在使用任何工具调用代码之前，需要先在部署服务时开启对应选项：

vLLM 部署需要同时添加--enable-auto-tool-choice与--tool-call-parser kimi_k2两个参数；
SGLang 部署需要添加--tool-call-parser kimi_k2参数；
如果使用的框架不在官方推荐列表（vLLM、SGLang、KTransformers、TensorRT-LLM）中，且没有现成的工具调用解析器，则需要使用本文后续介绍的"手动解析"方案。

完整的多节点启动命令（Tensor Parallelism、DP+EP 等不同并行策略）可参考部署指南。需要注意的是，推理引擎仍在频繁更新，部署参数应以各引擎官方主页为准，仓库中的命令仅为示例。

此外，Kimi K2 复用了DeepSeekV3CausalLM架构，并在config.json中设置"model_type": "kimi_k2"以便推理引擎将其与 DeepSeek-V3 区分并应用针对性优化。若遇到不在推荐列表中的框架，可临时将model_type改为"deepseek_v3"作为 workaround，但此时很可能需要手动解析工具调用。

Kimi K2 工具调用的完整流程

Kimi K2 的工具调用过程包含四个环节：

传递函数描述：将函数的结构化描述（JSON Schema）通过请求传给 Kimi K2；
模型决策并返回调用信息：Kimi K2 自主决定是否调用工具、调用哪个工具，并返回函数名与参数等必要信息；
用户执行调用并回填结果：由你的程序真正执行函数，收集返回结果，再以role='tool'的消息追加回对话历史；
模型基于结果继续生成：Kimi K2 依据函数执行结果继续生成内容，直到它认为已获得足够信息来回答用户问题。

在代码层面，第 3、4 步是一个循环：只要模型返回的finish_reason == "tool_calls"，就继续执行工具并回填结果，直到finish_reason不再是tool_calls。

准备工具：为 Kimi K2 声明函数

以一个实时查询天气的函数get_weather为例，它接收城市名参数并返回天气状况。为了让 Kimi K2 理解它的功能，需要准备一份结构化的 JSON 描述：

def get_weather(city): return {"weather": "Sunny"} # Collect the tool descriptions in tools tools = [{ "type": "function", "function": { "name": "get_weather", "description": "Get weather information. Call this tool when the user needs to get weather information", "parameters": { "type": "object", "required": ["city"], "properties": { "city": { "type": "string", "description": "City name", } } } } }] # Tool name->object mapping for easy calling later tool_map = { "get_weather": get_weather }

要点说明：

tools是一个列表，每个元素都遵循 OpenAI 的 function calling 约定：type固定为"function"，function内包含name、description和parameters（JSON Schema 格式）。required列出必填参数，properties逐一声明每个参数的类型与含义。
一个请求可以声明多个工具；描述写得越清晰，模型选择与传参就越准确。
tool_map用于把工具名映射到真实函数对象，方便循环中按名调用。注意 README 中tool_call_with_client的写法为tool_function(**tool_call_arguments)（解包关键字参数），而工具调用指南中的写法为tool_function(tool_call_arguments)（整体传参），两种方式均可，请与你的函数签名保持一致。

非流式对话：驱动模型自主决策并循环回填结果

使用openai.OpenAI客户端向 Kimi K2 发送消息与工具描述，模型会自主决定是否使用以及如何使用工具。若模型认为需要调用工具，返回结果的finish_reason将是'tool_calls'，其中包含工具调用信息；调用工具后，需要把工具结果追加到对话历史，再继续请求。整个过程可能重复多次，因此应循环检查finish_reason，直到其不再是tool_calls：

import json from openai import OpenAI model_name='moonshotai/Kimi-K2-Instruct' client = OpenAI(base_url=endpoint, api_key='xxx') messages = [ {"role": "user", "content": "What's the weather like in Beijing today? Let's check using the tool."} ] finish_reason = None while finish_reason is None or finish_reason == "tool_calls": completion = client.chat.completions.create( model=model_name, messages=messages, temperature=0.3, tools=tools, tool_choice="auto", ) choice = completion.choices[0] finish_reason = choice.finish_reason # Note: The finish_reason when tool calls end may vary across different engines, so this condition check needs to be adjusted accordingly if finish_reason == "tool_calls": messages.append(choice.message) for tool_call in choice.message.tool_calls: tool_call_name = tool_call.function.name tool_call_arguments = json.loads(tool_call.function.arguments) tool_function = tool_map[tool_call_name] tool_result = tool_function(tool_call_arguments) print("tool_result", tool_result) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_call_name, "content": json.dumps(tool_result), }) print('-' * 100) print(choice.message.content)

关键逻辑拆解：

循环条件：while finish_reason is None or finish_reason == "tool_calls"，首次请求时finish_reason为None必然进入循环；之后只要模型还在要求调用工具就继续迭代。
回填 assistant 消息：messages.append(choice.message)必须执行。模型返回的 assistant 消息里带有tool_calls信息，只有把它放回对话历史，后续请求中的工具结果才能与调用请求一一对应。
回填 tool 结果：用户执行工具得到的结果必须以role='tool'追加，并携带tool_call_id（与模型返回的调用 ID 对应）、name（工具名）以及content（工具结果，通常json.dumps序列化为字符串）。
参数解析：模型返回的tool_call.function.arguments是 JSON 字符串，需用json.loads解析为字典后再传给真实函数。
多工具并行：choice.message.tool_calls是一个列表，一次返回可能包含多个工具调用，循环内逐个执行并回填即可。
引擎差异注意：不同推理引擎对"工具调用结束"的finish_reason取值可能不同，这段条件判断需要按实际引擎调整。

需要注意的是，工具调用指南示例中temperature=0.3，而 README 中给出的 Kimi-K2-Instruct 推荐温度为0.6，两者均可按场景调整。

流式模式下的工具调用

流式（stream=True）模式下，模型输出被切成多个 chunk 返回，工具调用信息也会被分片。因此需要按index把每个工具调用的id、function.name、function.arguments片段累积拼接，直到拿到一个完整的工具调用：

messages = [ {"role": "user", "content": "What's the weather like in Beijing today? Let's check using the tool."} ] finish_reason = None msg = '' while finish_reason is None or finish_reason == "tool_calls": completion = client.chat.completions.create( model=model_name, messages=messages, temperature=0.3, tools=tools, tool_choice="auto", stream=True ) tool_calls = [] for chunk in completion: delta = chunk.choices[0].delta if delta.content: msg += delta.content if delta.tool_calls: for tool_call_chunk in delta.tool_calls: if tool_call_chunk.index is not None: # Extend the tool_calls list while len(tool_calls) <= tool_call_chunk.index: tool_calls.append({ "id": "", "type": "function", "function": { "name": "", "arguments": "" } }) tc = tool_calls[tool_call_chunk.index] if tool_call_chunk.id: tc["id"] += tool_call_chunk.id if tool_call_chunk.function.name: tc["function"]["name"] += tool_call_chunk.function.name if tool_call_chunk.function.arguments: tc["function"]["arguments"] += tool_call_chunk.function.arguments finish_reason = chunk.choices[0].finish_reason # Note: The finish_reason when tool calls end may vary across different engines, so this condition check needs to be adjusted accordingly if finish_reason == "tool_calls": for tool_call in tool_calls: tool_call_name = tool_call['function']['name'] tool_call_arguments = json.loads(tool_call['function']['arguments']) tool_function = tool_map[tool_call_name] tool_result = tool_function(tool_call_arguments) messages.append({ "role": "tool", "tool_call_id": tool_call['id'], "name": tool_call_name, "content": json.dumps(tool_result), }) # The text generated by the tool call is not the final version, reset msg msg = '' print(msg)

流式场景的几个要点：

按 index 归位：tool_call_chunk.index标识该分片属于第几个工具调用。代码用while len(tool_calls) <= tool_call_chunk.index先把占位元素补齐，再通过tc = tool_calls[tool_call_chunk.index]定位目标，用+=拼接分片内容。
内容与调用信息分路收集：普通文本流式累积到msg；工具调用片段累积到tool_calls列表。
重置 msg：当finish_reason == "tool_calls"时，流式过程中生成的那段文本只是中间产物（通常是模型对调用动作的叙述），并非最终答案，因此执行完工具后要把msg清空，下一轮继续累积。
结束条件与回填：循环直到finish_reason不再是tool_calls；每轮工具执行结果仍以role='tool'追加到messages。

手动解析工具调用：无解析器框架下的兜底方案

当使用的推理服务没有提供工具调用解析器时，可以完全绕过 OpenAI 兼容的 tools 参数，直接调用原生 completions 接口并手动解析模型输出。Kimi K2 的工具调用请求遵循一套内置的标记（token）协议：

整个工具调用段落被<|tool_calls_section_begin|>和<|tool_calls_section_end|>包裹；
每个工具调用被<|tool_call_begin|>和<|tool_call_end|>包裹；
工具 ID 与参数之间以<|tool_call_argument_begin|>分隔；
工具 ID 的格式为functions.{func_name}:{idx}，例如functions.get_weather:0，从中可以直接解析出函数名。

基于以上规则，可以用requests直接向 completions 接口发请求，并用AutoTokenizer的apply_chat_template把tools注入对话模板：

import requests from transformers import AutoTokenizer messages = [ {"role": "user", "content": "What's the weather like in Beijing today? Let's check using the tool."} ] msg = '' tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) while True: text = tokenizer.apply_chat_template( messages, tokenize=False, tools=tools, add_generation_prompt=True, ) payload = { "model": model_name, "prompt": text, "max_tokens": 512 } response = requests.post( f"{endpoint}/completions", headers={"Content-Type": "application/json"}, json=payload, stream=False, ) raw_out = response.json() raw_output = raw_out["choices"][0]["text"] tool_calls = extract_tool_call_info(raw_output) if len(tool_calls) == 0: # No tool calls msg = raw_output break else: for tool_call in tool_calls: tool_call_name = tool_call['function']['name'] tool_call_arguments = json.loads(tool_call['function']['arguments']) tool_function = tool_map[tool_call_name] tool_result = tool_function(tool_call_arguments) messages.append({ "role": "tool", "tool_call_id": tool_call['id'], "name": tool_call_name, "content": json.dumps(tool_result), }) print('-' * 100) print(msg)

这段代码与前面流程的差异在于：

用tokenizer.apply_chat_template(messages, tokenize=False, tools=tools, add_generation_prompt=True)把对话历史与工具描述渲染成模型输入文本，tools会被注入到模板中的工具声明区；
直接请求{endpoint}/completions原生接口，从choices[0]["text"]拿到原始输出；
用extract_tool_call_info解析原始输出；若没有工具调用则直接输出结果并退出，否则执行工具、回填role='tool'消息后继续循环；
max_tokens: 512是单次生成的上限，可按需调整。

这里extract_tool_call_info负责解析模型输出并返回调用信息，一个简单的实现如下：

def extract_tool_call_info(tool_call_rsp: str): if '<|tool_calls_section_begin|>' not in tool_call_rsp: # No tool calls return [] import re pattern = r"<\|tool_calls_section_begin\|>(.*?)<\|tool_calls_section_end\|>" tool_calls_sections = re.findall(pattern, tool_call_rsp, re.DOTALL) # Extract multiple tool calls func_call_pattern = r"<\|tool_call_begin\|>\s*(?P<tool_call_id>[\w\.]+:\d+)\s*<\|tool_call_argument_begin\|>\s*(?P<function_arguments>.*?)\s*<\|tool_call_end\|>" tool_calls = [] for match in re.findall(func_call_pattern, tool_calls_sections[0], re.DOTALL): function_id, function_args = match # function_id: functions.get_weather:0 function_name = function_id.split('.')[1].split(':')[0] tool_calls.append( { "id": function_id, "type": "function", "function": { "name": function_name, "arguments": function_args } } ) return tool_calls

解析逻辑说明：

先用'<|tool_calls_section_begin|>' not in tool_call_rsp快速判断输出中是否存在工具调用，不存在直接返回空列表；
第一层正则提取<|tool_calls_section_begin|>与<|tool_calls_section_end|>之间的整个工具调用段（re.DOTALL让.能匹配换行）；
第二层正则按<|tool_call_begin|>、<|tool_call_argument_begin|>、<|tool_call_end|>的边界切出每个调用：tool_call_id形如functions.get_weather:0，function_arguments为 JSON 参数字符串；
通过function_id.split('.')[1].split(':')[0]从 ID 中还原函数名，例如functions.get_weather:0→get_weather；
最终统一输出与 OpenAI 兼容结构一致的结果（id/type/function.name/function.arguments），方便复用同一套后续执行逻辑。

部署侧参数与工具调用的联动

README 明确指出，工具调用管线要求推理引擎支持 Kimi K2 的原生工具解析逻辑。从部署指南可以看到，无论采用哪种引擎，工具调用都需要显式开启：

推理引擎	启用工具调用所需参数
vLLM	`--enable-auto-tool-choice --tool-call-parser kimi_k2`
SGLang	`--tool-call-parser kimi_k2`
其他框架（无解析器）	需手动解析（使用上文`extract_tool_call_info`方案），必要时临时将`config.json`的`model_type`改为`"deepseek_v3"`

vLLM 以 Tensor Parallel 方式启动服务的最小示例：

vllm serve $MODEL_PATH \ --port 8000 \ --served-model-name kimi-k2 \ --trust-remote-code \ --tensor-parallel-size 16 \ --enable-auto-tool-choice \ --tool-call-parser kimi_k2

其中--enable-auto-tool-choice与--tool-call-parser kimi_k2是启用工具使用时的必选项；--tensor-parallel-size 16表示 16 卡张量并行，若 GPU 数超过 16 需结合流水线并行使用。SGLang 同样只需添加--tool-call-parser kimi_k2即可。更大的并行规模（如 DP+EP）与 Prefill-Decode 分离部署的完整命令，均见部署指南。

实践建议与注意事项

先确认引擎支持解析器：非流式/流式方案依赖引擎内置的kimi_k2解析器；只有确认服务端已开启（vLLM 的--enable-auto-tool-choice --tool-call-parser kimi_k2或 SGLang 的--tool-call-parser kimi_k2）才能直接使用 OpenAI 兼容写法，否则请使用手动解析方案。
finish_reason因引擎而异：两处代码都保留了注释提醒——不同引擎对工具调用结束的finish_reason取值可能不同，切换引擎时需相应调整判断条件。
消息顺序不可错乱：assistant 的tool_calls消息必须紧跟在用户消息之后、工具结果之前；每个role='tool'消息必须携带与之配对的tool_call_id，否则模型无法正确关联调用与结果。
支持多轮、多工具：一次返回可能包含多个tool_calls，循环内逐个执行即可；模型也可能在获得结果后再次发起新一轮工具调用，外层while循环负责处理这种多轮交互，直到模型给出最终答复。
工具结果建议 JSON 序列化：将工具返回结果用json.dumps转成字符串放入content，是官方示例的统一做法，有助于模型稳定解析。
手动解析的适用场景：extract_tool_call_info适用于任何能拿到原始生成文本的服务（原生 completions 接口、本地推理等），是框架没有内置解析器时的可靠兜底方案。