from typing import *
import json
import httpx
from openai import OpenAI
client = OpenAI(
api_key="MOONSHOT_API_KEY", # Replace MOONSHOT_API_KEY with the API Key you obtained from the Kimi Open Platform
base_url="https://api.moonshot.ai/v1",
)
functions = [
{
"name": "search", # The name of the function, please use English letters (uppercase and lowercase), numbers, plus hyphens and underscores as the function name
"description": """
Search for content on the internet using a search engine.
Call this tool when your knowledge cannot answer the user's question or when the user requests you to perform an online search. Extract the content the user wants to search for from the conversation with the user and use it as the value of the query parameter.
The search results include the title of the website, the website's address (URL), and a brief introduction to the website.
""", # Description of the function, write the specific function here and the usage scenario so that the Kimi large language model can correctly choose which functions to use
"parameters": { # Use the parameters field to define the parameters accepted by the function
"type": "object", # Always use type: object to make the Kimi large language model generate a JSON Object parameter
"required": ["query"], # Use the required field to tell the Kimi large language model which parameters are required
"properties": { # Properties contain the specific parameter definitions, and you can define multiple parameters
"query": { # Here, the key is the parameter name, and the value is the specific definition of the parameter
"type": "string", # Use type to define the parameter type
"description": """
The content the user wants to search for, extract it from the user's question or chat context.
""" # Use description to describe the parameter so that the Kimi large language model can better generate the parameter
}
}
}
}
]
def search_impl(query: str) -> List[Dict[str, Any]]:
"""
search_impl uses a search engine to search for query. Most mainstream search engines (such as Bing) provide API calls, and you can choose the one you like.
You can call the search engine API of your choice and place the website title, website link, and website introduction information in a dict and return it.
This is just a simple example, and you may need to write some authentication, validation, and parsing code.
"""
r = httpx.get("https://your.search.api", params={"query": query})
return r.json()
def search(arguments: Dict[str, Any]) -> Any:
query = arguments["query"]
result = search_impl(query)
return {"result": result}
function_map = {
"search": search,
}
# ==========================================================================================================================================================
# Tools are a superset of functions, so we can construct tools using the already defined functions. We loop through each function and create the corresponding tool format;
# At the same time, we also generate the corresponding tool_map.
# ==========================================================================================================================================================
tools = []
tool_map = {}
for function in functions:
tool = {
"type": "function",
"function": function,
}
tools.append(tool)
tool_map[function["name"]] = function_map[function["name"]]
messages = [
{"role": "system",
"content": "You are Kimi, an artificial intelligence assistant provided by Moonshot AI. You are more proficient in Chinese and English conversations. You provide users with safe, helpful, and accurate answers. You also refuse to answer any questions involving terrorism, racism, pornography, or violence. Moonshot AI is a proper noun and should not be translated into other languages."},
{"role": "user", "content": "Please search the internet for Context Caching and tell me what it is."} # The user asks Kimi to search online
]
finish_reason = None
# ==========================================================================================================================================================
# Here, we change the finish_reason value check from function_call to tool_calls
# ==========================================================================================================================================================
# while finish_reason is None or finish_reason == "function_call":
while finish_reason is None or finish_reason == "tool_calls":
completion = client.chat.completions.create(
model="kimi-k2.5",
messages=messages,
# ==========================================================================================================================================================
# We no longer use the functions parameter, but instead use the tools parameter to enable tool calls
# ==========================================================================================================================================================
# function=functions,
tools=tools, # <-- We submit the defined tools to Kimi via the tools parameter
)
choice = completion.choices[0]
finish_reason = choice.finish_reason
# ==========================================================================================================================================================
# Here, we replace the original function_call execution logic with the tool_calls execution logic;
# Note that since there may be multiple tool_calls, we need to execute each one using a for loop.
# ==========================================================================================================================================================
# if finish_reason == "function_call":
# messages.append(choice.message)
# function_call_name = choice.message.function_call.name
# function_call_arguments = json.loads(choice.message.function_call.arguments)
# function_call = function_map[function_call_name]
# function_result = function_call(function_call_arguments)
# messages.append({
# "role": "function",
# "name": function_call_name,
# "content": json.dumps(function_result)
# })
if finish_reason == "tool_calls": # <-- Check if the response contains tool_calls
messages.append(choice.message) # <-- Add the assistant message from Kimi to the context for the next request
for tool_call in choice.message.tool_calls: # <-- Loop through each tool_call as there may be multiple
tool_call_name = tool_call.function.name
tool_call_arguments = json.loads(tool_call.function.arguments) # <-- The arguments are serialized JSON, so we need to deserialize them
tool_function = tool_map[tool_call_name] # <-- Use tool_map to quickly find the function to execute
tool_result = tool_function(tool_call_arguments)
# Construct a message with role=tool to show the result of the tool call to the model;
# Note that we need to provide the tool_call_id and name fields in the message so that Kimi can
# correctly match it to the corresponding tool_call.
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call_name,
"content": json.dumps(tool_result), # <-- We agree to submit the tool call result as a string, so we serialize it here
})
print(choice.message.content) # <-- Finally, we return the model's response to the user
