import jsonimport osfrom datetime import datefrom openai import OpenAI
# -------- 1. Tool implementations --------def get_weather(location: str, unit: str = "celsius") -> str: """Get current weather information for a city (simulated).""" table = { "London": {"temperature": 15, "condition": "sunny", "humidity": 45}, "New York": {"temperature": 20, "condition": "cloudy", "humidity": 60}, "Tokyo": {"temperature": 25, "condition": "rainy", "humidity": 75}, } w = dict(table.get(location, {"temperature": 18, "condition": "unknown", "humidity": 50})) if unit == "fahrenheit": w["temperature"] = w["temperature"] * 9 / 5 + 32 w["unit"] = "°F" else: w["unit"] = "°C" return json.dumps(w, ensure_ascii=False)
def calculate(expression: str) -> str: """Evaluate a Python-style arithmetic expression.""" try: return json.dumps({"expression": expression, "result": eval(expression)}, ensure_ascii=False) except Exception as e: return json.dumps({"expression": expression, "error": str(e)}, ensure_ascii=False)
available_tools = {"get_weather": get_weather, "calculate": calculate}
# -------- 2. Tool schemas (OpenAI function-calling format) --------tools = [ { "type": "function", "function": { "name": "get_weather", "description": "Get current weather information for a city.", "parameters": { "type": "object", "properties": { "location": {"type": "string", "description": "City name, e.g. 'London'."}, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "Temperature unit (default: celsius).", }, }, "required": ["location"], }, }, }, { "type": "function", "function": { "name": "calculate", "description": "Evaluate a Python-style arithmetic expression.", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "Expression to evaluate, e.g. '(25 + 15) * 3 - 10'.", }, }, "required": ["expression"], }, }, },]
# -------- 3. System prompt --------SYSTEM_PROMPT = f"""You are Apodex, an AI assistant developed by Apodex AI.
Apodex is the flagship agent of Apodex AI. Rather than a conventional conversational LLM, it is a general-purpose solver designed for mission-critical tasks.
Current time: {date.today()}. In this environment you have access to a set of tools you can use to answer the user's question.
You only have access to the tools provided. You can use multiple tools per message, and will receive the results of those tools in the user's next response. You use tools step-by-step to accomplish a given task.
# General Objective
You accomplish a given task iteratively, breaking it down into clear steps and working through them methodically."""
# -------- 4. Agentic loop --------def run_agent(user_query: str, model: str = "apodex/Apodex-1.1-mini-GPTQ-Int4", max_turns: int = 20): client = OpenAI( api_key=os.environ.get("OPENAI_API_KEY", "EMPTY"), base_url=os.environ.get("BASE_URL", "<http://localhost:1234/v1>"), )
messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_query}, ] print(f"\\n{'=' * 60}\\nUser: {user_query}\\n{'=' * 60}\\n")
for turn in range(max_turns): resp = client.chat.completions.create( model=model, messages=messages, tools=tools, parallel_tool_calls=True, temperature=1.0, top_p=0.95, max_tokens=16384, extra_body={"repetition_penalty": 1.05}, ) msg = resp.choices[0].message
# Optional: print reasoning if the server exposes it (qwen3 reasoning parser) reasoning = getattr(msg, "reasoning_content", None) if reasoning: print(f"[think] {reasoning.strip()}\\n") if msg.content: print(f"[assistant] {msg.content.strip()}\\n")
messages.append(msg)
# No more tool calls -> final answer if not msg.tool_calls: print(f"💬 Final answer:\\n{msg.content}\\n") return msg.content
# Execute every tool call requested in this turn for call in msg.tool_calls: name = call.function.name args = json.loads(call.function.arguments or "{}") print(f"🔧 call {name}({args})") try: result = available_toolsname except Exception as e: result = json.dumps({"error": f"{type(e).__name__}: {e}"}, ensure_ascii=False) print(f" ↳ {result}\\n") messages.append({ "role": "tool", "tool_call_id": call.id, "content": result, })
print("⚠️ Reached max_turns without a final answer.") return None
if __name__ == "__main__": run_agent("What's the weather in London in Fahrenheit, and what's (25 + 15) * 3 - 10?")