Kimi K2 使用指南 2026：Agent 开发与 Tool Calling 实战

from openai import OpenAI

client = OpenAI(
    api_key="your-crazyrouter-key",
    base_url="https://crazyrouter.com/v1"
)

response = client.chat.completions.create(
    model="kimi-k2",
    messages=[
        {"role": "system", "content": "你是一个专业的技术助手。"},
        {"role": "user", "content": "解释 Python 中的装饰器模式，并给出实际应用示例。"}
    ],
    temperature=0.7,
    max_tokens=2048
)

print(response.choices[0].message.content)

curl https://crazyrouter.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-crazyrouter-key" \
  -d '{
    "model": "kimi-k2",
    "messages": [
      {"role": "user", "content": "用 Python 实现一个简单的 LRU 缓存"}
    ]
  }'

from openai import OpenAI
import json

client = OpenAI(
    api_key="your-crazyrouter-key",
    base_url="https://crazyrouter.com/v1"
)

# 定义工具列表
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 '北京'、'上海'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_forecast",
            "description": "获取指定城市未来几天的天气预报",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "days": {"type": "integer", "description": "预报天数，1-7"}
                },
                "required": ["city", "days"]
            }
        }
    }
]

def execute_tool(tool_name: str, arguments: dict) -> str:
    """模拟工具执行（实际项目中替换为真实 API 调用）"""
    if tool_name == "get_weather":
        return json.dumps({
            "city": arguments["city"],
            "temperature": 22,
            "condition": "晴",
            "humidity": 45
        })
    elif tool_name == "get_forecast":
        forecasts = []
        for i in range(arguments.get("days", 3)):
            forecasts.append({
                "day": f"第{i+1}天",
                "high": 25 - i,
                "low": 15 - i,
                "condition": ["晴", "多云", "小雨"][i % 3]
            })
        return json.dumps(forecasts)
    return json.dumps({"error": "未知工具"})


def run_agent(user_query: str):
    """运行 Agent 主循环"""
    messages = [
        {"role": "system", "content": "你是一个天气助手，可以查询天气和预报。请用中文回答。"},
        {"role": "user", "content": user_query}
    ]
    
    while True:
        response = client.chat.completions.create(
            model="kimi-k2",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        messages.append(message)
        
        # 如果没有工具调用，返回最终结果
        if not message.tool_calls:
            return message.content
        
        # 执行所有工具调用（Kimi K2 支持并行调用）
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            print(f"  🔧 调用工具: {func_name}({func_args})")
            result = execute_tool(func_name, func_args)
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

# 测试
answer = run_agent("北京今天天气怎么样？顺便告诉我未来3天的预报")
print(answer)

  🔧 调用工具: get_weather({"city": "北京"})
  🔧 调用工具: get_forecast({"city": "北京", "days": 3})

北京今天天气晴朗，气温22°C，湿度45%。

未来3天预报：
- 第1天：晴，最高25°C / 最低15°C
- 第2天：多云，最高24°C / 最低14°C  
- 第3天：小雨，最高23°C / 最低13°C

建议第3天出门带伞哦！

import json
from openai import OpenAI
from typing import Callable

class KimiK2Agent:
    def __init__(self, api_key: str, system_prompt: str):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://crazyrouter.com/v1"
        )
        self.system_prompt = system_prompt
        self.tools = []
        self.tool_handlers = {}
    
    def register_tool(self, name: str, description: str, 
                      parameters: dict, handler: Callable):
        """注册一个工具"""
        self.tools.append({
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": parameters
            }
        })
        self.tool_handlers[name] = handler
    
    def run(self, query: str, max_iterations: int = 10) -> str:
        """运行 Agent"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": query}
        ]
        
        for _ in range(max_iterations):
            response = self.client.chat.completions.create(
                model="kimi-k2",
                messages=messages,
                tools=self.tools if self.tools else None,
                tool_choice="auto" if self.tools else None
            )
            
            msg = response.choices[0].message
            messages.append(msg)
            
            if not msg.tool_calls:
                return msg.content
            
            for tc in msg.tool_calls:
                handler = self.tool_handlers.get(tc.function.name)
                if handler:
                    args = json.loads(tc.function.arguments)
                    result = handler(**args)
                else:
                    result = {"error": f"未注册的工具: {tc.function.name}"}
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": json.dumps(result, ensure_ascii=False)
                })
        
        return "达到最大迭代次数，Agent 停止运行。"


# 使用示例
agent = KimiK2Agent(
    api_key="your-crazyrouter-key",
    system_prompt="你是一个全能助手，可以搜索网页、查询数据库和发送通知。"
)

# 注册工具
agent.register_tool(
    name="search_web",
    description="搜索互联网获取最新信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"}
        },
        "required": ["query"]
    },
    handler=lambda query: {"results": [f"搜索结果: {query}"]}
)

result = agent.run("帮我搜索最新的 Python 3.14 新特性")
print(result)