Kimi Agent의 구조와 실제 성능 검증

import os
import json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["MOONSHOT_API_KEY"],
    base_url="https://api.moonshot.ai/v1",
)

MODEL = "kimi-k3"
TOOLS = [{
    "type": "function",
    "function": {
        "name": "count_words",
        "description": "Counts the number of words in a block of text.",
        "parameters": {
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        },
    },
}]

def count_words(text: str) -> int:
    return len(text.split())

def run_task(task: str, max_turns: int = 6) -> dict:
    """Runs a task through Kimi K3, executing any tool calls it makes, and returns a small report of what happened."""
    messages = [{"role": "user", "content": task}]
    total_prompt_tokens = 0
    total_completion_tokens = 0
    turns_used = 0
    for turn in range(max_turns):
        turns_used = turn + 1
        response = client.chat.completions.create(
            model=MODEL,
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
            reasoning_effort="max", # K3 replaced the old `thinking` param with this
        )
        usage = response.usage
        total_prompt_tokens += usage.prompt_tokens
        total_completion_tokens += usage.completion_tokens
        message = response.choices[0].message
        if response.choices[0].finish_reason != "tool_calls":
            return {
                "answer": message.content,
                "turns_used": turns_used,
                "prompt_tokens": total_prompt_tokens,
                "completion_tokens": total_completion_tokens,
            }
        messages.append(message.model_dump(exclude_none=True))
        for tool_call in message.tool_calls:
            if tool_call.function.name == "count_words":
                args = json.loads(tool_call.function.arguments)
                result = count_words(args["text"])
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })
    return {"answer": None, "turns_used": turns_used, "error": "Hit max_turns without finishing"}