Kimi Agent Build

상위 문서: ← Kimi Core workflows 출처: Kimi API Platform 공식 문서 — Build an Agent with Kimi K3 한글화 (2026-07-22) 공급사: Moonshot AI (月之暗面)

Kimi K3는 복잡한 작업을 위한 추론·코딩·도구 호출 능력을 제공합니다. 이 가이드는 공식 web-search 도구와 커스텀 도구 하나를 결합한 산업 리서치 에이전트 예제로, 실행 가능한 bounded 에이전트 구축법을 보여줍니다.

1. 작업 분해

산업 리서치를 3단계로 분해:

1. 수집 (Retrieve): 리서치 범위 정의, 최신 데이터·기업 정보·뉴스 검색
2. 분석 (Analyze): 소스 비교, 충돌 식별, 사실·추정·추론 구분
3. 전달 (Deliver): 요약·핵심 발견·리스크·소스 목록이 포함된 구조화된 리포트

이 분해로 모델이 도구로 검색·결정적 작업을 처리하는 동안, 모델 자체는 계획과 판단에 집중할 수 있음.

2. 도구 설계

두 종류의 도구 결합:

• 공식 web-search: 최신 산업 소스 검색. 플랫폼은 fetch, code-runner, excel 등도 제공. 전체 목록과 Formula API 흐름은 Official Tools 참고
• 커스텀 build_research_plan: 주제·지역·연도 범위에서 결정론적 스코프 생성. 로컬 함수 도구 선언·실행 예시

커스텀 도구는 JSON Schema로 인자를 기술. additionalProperties: false + required로 필수 필드 명시 → 잘못된 인자 감소.

3. 시스템 프롬프트

역할·워크플로우·품질 경계에 집중. 인자 상세는 도구 스키마에 두기:

SYSTEM_PROMPT = """You are an industry research assistant. Define the scope first, retrieve and cross-check information, then write a concise report.
Requirements:
- Separate confirmed facts, estimates, and inferences; support key findings with multiple sources where possible.
- Never fabricate data or sources; state the search scope and gaps when evidence is insufficient.
- Include an executive summary, key findings, risks and limitations, and a source list.
- Answer in the same language as the user.
"""

비즈니스 포맷·컴플라이언스·청중이 바뀔 때 제약을 점진적으로 추가. 자세한 내용은 Prompt Best Practices 참고.

4. K3 API 설정

• Python 3.9+
• openai + httpx 설치
• MOONSHOT_API_KEY 환경변수
• Global endpoint: https://api.moonshot.ai/v1, model: kimi-k3

K3는 항상 사고하며, reasoning_effort는 "low" / "high" / "max" (기본 "max") 지원. 도구 루프는 SDK가 반환한 완전한 assistant 메시지를 messages에 추가해야 함 — content/tool_calls만 복사하면 reasoning_content 손실.

5. 완전한 에이전트 루프

agent.py로 저장. web-search 선언을 동적으로 로드, 커스텀 도구는 로컬에서 실행, 최대 8라운드로 도구 호출 처리.

import asyncio
import json
import os
import httpx
from openai import AsyncOpenAI

BASE_URL = "https://api.moonshot.ai/v1"
MODEL = "kimi-k3"
MAX_TOOL_ROUNDS = 8

RESEARCH_PLAN_TOOL = {
    "type": "function",
    "function": {
        "name": "build_research_plan",
        "description": "Build a research plan from an industry topic, region, and time range",
        "parameters": {
            "type": "object",
            "properties": {
                "topic": {"type": "string", "description": "Industry or subject to research"},
                "region": {"type": "string", "enum": ["China", "Global", "United States", "Europe"]},
                "start_year": {"type": "integer"},
                "end_year": {"type": "integer"},
            },
            "required": ["topic", "region", "start_year", "end_year"],
            "additionalProperties": False,
        },
    },
}


def build_research_plan(topic: str, region: str, start_year: int, end_year: int) -> str:
    if start_year > end_year:
        return json.dumps({"error": "start_year must not exceed end_year"})
    plan = {
        "topic": topic, "region": region, "period": f"{start_year}-{end_year}",
        "dimensions": ["market size and growth", "value chain and major companies",
                       "technology trends", "policy and risks"],
        "search_queries": [
            f"{region} {topic} market size {start_year} {end_year}",
            f"{region} {topic} major companies technology trends",
            f"{region} {topic} policy risks",
        ],
    }
    return json.dumps(plan)


class IndustryResearchAgent:
    def __init__(self):
        api_key = os.environ["MOONSHOT_API_KEY"]
        self.openai = AsyncOpenAI(api_key=api_key, base_url=BASE_URL)
        self.http = httpx.AsyncClient(base_url=BASE_URL, headers={"Authorization": f"Bearer {api_key}"}, timeout=60.0)

    async def load_formula(self, formula_uri):
        r = await self.http.get(f"/formulas/{formula_uri}/tools")
        r.raise_for_status()
        return r.json().get("tools", [])

    async def call_formula(self, formula_uri, name, args):
        r = await self.http.post(f"/formulas/{formula_uri}/fibers", json={"name": name, "arguments": json.dumps(args)})
        r.raise_for_status()
        fiber = r.json()
        ctx = fiber.get("context", {})
        result = ctx.get("output") or ctx.get("encrypted_output") or ""
        if fiber.get("status") != "succeeded":
            result = fiber.get("error") or ctx.get("error") or "Unknown tool error"
        return result if isinstance(result, str) else json.dumps(result)

    async def run(self, question):
        official_tools = await self.load_formula("moonshot/web-search:latest")
        tool_to_formula = {t["function"]["name"]: "moonshot/web-search:latest" for t in official_tools}
        tools = [RESEARCH_PLAN_TOOL, *official_tools]
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": question},
        ]
        for _ in range(MAX_TOOL_ROUNDS):
            response = await self.openai.chat.completions.create(model=MODEL, messages=messages, tools=tools, max_completion_tokens=8192)
            message = response.choices[0].message
            messages.append(message)  # <-- 핵심: assistant 메시지 전체를 보존
            if response.choices[0].finish_reason == "length":
                raise RuntimeError("출력 잘림 — max_completion_tokens 증가 또는 결과 단축")
            if not message.tool_calls:
                return message.content
            for tc in message.tool_calls:
                name = tc.function.name
                try:
                    args = json.loads(tc.function.arguments or "{}")
                    if name == "build_research_plan":
                        result = build_research_plan(**args)
                    elif name in tool_to_formula:
                        result = await self.call_formula(tool_to_formula[name], name, args)
                    else:
                        raise ValueError(f"Unknown tool: {name}")
                except Exception as e:
                    result = json.dumps({"error": f"{type(e).__name__}: {e}"})
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
        raise RuntimeError(f"Tool calls exceeded {MAX_TOOL_ROUNDS} rounds")


async def main():
    agent = IndustryResearchAgent()
    try:
        report = await agent.run("Research the development of China's humanoid-robot industry from 2024 to 2026")
        print(report)
    finally:
        await agent.http.aclose()
        await agent.openai.close()


if __name__ == "__main__":
    asyncio.run(main())

6. 실행 및 트러블슈팅

python3 agent.py

일반적인 문제:

• finish_reason="length" → max_completion_tokens 늘리기 (또는 도구 결과 단축)
• 도구 라운드 한도 도달 → 도구 정의 중복·모호한 결과 확인 후 작업 범위 좁히기
• 도구 인자 에러 반복 → function schema의 required / enum / additionalProperties 확인
• 후속 도구 라운드 실패 → assistant 메시지 전체 + tool_call_id 정확히 보존하는지 확인
• 공식 도구 요청 실패 → 엔드포인트·API 키·Formula URI·도구 가용성 확인

7. 커스텀 도구와 최적화

build_research_plan 예제는 전체 경로를 다룸 — 스키마 선언, 로컬 디스패치, JSON 반환, tool_call_id 매칭. DB·내부 검색·파일 생성기를 연결하려면 함수 구현만 교체하고 스키마/반환 형식은 유지.

추가 최적화:

• 현재 작업에 필요한 도구만 노출, 중복 도구 회피
• 도구 입력 검증 + 구조화된 에러 반환 → 모델이 인자 수정 가능
• 대규모 인벤토리에는 Dynamically Loaded Tools
• tool_choice / reasoning_effort 같은 제어 사용 전 Kimi K3 Tool Calling Best Practices 참고

---

변경 이력

날짜	변경
2026-07-22	초판 작성 (공식 Agent Build 가이드 한글화)
2026-07-22	제목에 "Kimi" 접두사 추가, 본문 H1 중복 제거