Kimi K2 Documentation

When using stream=True, tool calls are delivered in chunks. You must accumulate the delta.tool_calls fragments (specifically id, function.name, and function.arguments) into a list until the stream finishes.

Important: After processing all tool calls and appending the results to messages, you must reset the accumulated text message (msg = '') because the text generated during the tool-calling phase is not the final response.

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:
                    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
    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),
            })
        msg = ''

    print(msg)

If your service does not provide a built-in tool-call parser, you can manually extract tool calls from the model's raw text output using specific delimiters.

Delimiters:

• Section start/end: <|tool_calls_section_begin|> and <|tool_calls_section_end|>
• Individual tool call start/end: <|tool_call_begin|> and <|tool_call_end|>
• Argument separator: <|tool_call_argument_begin|>

ID Format: The tool ID follows the pattern functions.{func_name}:{idx}. You can extract the function name by splitting this string.

Implementation Example:

def extract_tool_call_info(tool_call_rsp: str):
    if '<|tool_calls_section_begin|>' not in tool_call_rsp:
        return []
    import re
    pattern = r"<\|tool_calls_section_begin\|>(.*?)<\|tool_calls_section_end\|>"
    
    tool_calls_sections = re.findall(pattern, tool_call_rsp, re.DOTALL)
    
    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_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

To use tool calling, you must provide a structured description of your functions in a tools list. Each tool should be an object of type: "function" containing a function object with name, description, and parameters (following JSON Schema format). It is recommended to maintain a tool_map that maps function names to their actual Python implementations for easy execution during the tool-calling loop.

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
}

You can use the openai.OpenAI client to interact with Kimi-K2. When the model decides to use a tool, it returns a finish_reason='tool_calls'.

Workflow:

1. Send messages and the tools list to the model.
2. Check if finish_reason == "tool_calls".
3. If true, iterate through choice.message.tool_calls, execute the local functions using your tool_map, and append the results to the messages list with role='tool'.
4. Include the tool_call_id, name, and the stringified JSON content of the result.
5. Repeat the loop until finish_reason is no longer 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
    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)