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
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.
"""
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"],
"description": "Region to research",
},
"start_year": {
"type": "integer",
"description": "First year in the research period",
},
"end_year": {
"type": "integer",
"description": "Last year in the research period",
},
},
"required": ["topic", "region", "start_year", "end_year"],
"additionalProperties": False,
},
},
}
def build_research_plan(
topic: str, region: str, start_year: int, end_year: int
) -> str:
"""Build a deterministic scope as a minimal custom-tool example."""
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) -> None:
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: str
) -> tuple[list[dict], dict[str, str]]:
response = await self.http.get(f"/formulas/{formula_uri}/tools")
response.raise_for_status()
tools = response.json().get("tools", [])
if not tools:
raise RuntimeError(f"Formula {formula_uri} returned no tools")
tool_to_formula = {
tool["function"]["name"]: formula_uri
for tool in tools
if tool.get("type") == "function" and tool.get("function")
}
if not tool_to_formula:
raise RuntimeError(f"Formula {formula_uri} returned no callable function tools")
return tools, tool_to_formula
async def call_formula(
self, formula_uri: str, name: str, arguments: dict
) -> str:
response = await self.http.post(
f"/formulas/{formula_uri}/fibers",
json={"name": name, "arguments": json.dumps(arguments)},
)
response.raise_for_status()
fiber = response.json()
context = fiber.get("context", {})
if fiber.get("status") == "succeeded":
result = context.get("output") or context.get("encrypted_output") or ""
else:
result = fiber.get("error") or context.get("error") or "Unknown tool error"
if isinstance(result, str):
return result
return json.dumps(result)
async def run(self, question: str) -> str:
official_tools, tool_to_formula = await self.load_formula(
"moonshot/web-search:latest"
)
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,
)
choice = response.choices[0]
message = choice.message
# Append the complete SDK message, preserving reasoning_content and tool_calls.
messages.append(message)
if choice.finish_reason == "length":
raise RuntimeError(
"The model output was truncated by max_completion_tokens; "
"increase the limit or shorten tool results"
)
if not message.tool_calls:
if choice.finish_reason != "stop":
raise RuntimeError(
f"Unexpected finish reason: {choice.finish_reason}"
)
if not message.content:
raise RuntimeError("The model returned no final report")
return message.content
for tool_call in message.tool_calls:
name = tool_call.function.name
try:
arguments = json.loads(tool_call.function.arguments or "{}")
if not isinstance(arguments, dict):
raise ValueError("Tool arguments must be a JSON object")
if name == "build_research_plan":
result = build_research_plan(**arguments)
elif name in tool_to_formula:
result = await self.call_formula(
tool_to_formula[name], name, arguments
)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as exc:
result = json.dumps(
{"error": f"{type(exc).__name__}: {exc}"}
)
# Return every result and continue other calls even if one tool fails.
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
}
)
raise RuntimeError(f"Tool calls exceeded {MAX_TOOL_ROUNDS} rounds")
async def close(self) -> None:
try:
await self.openai.close()
finally:
await self.http.aclose()
async def main() -> None:
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.close()
if __name__ == "__main__":
asyncio.run(main())
