#!/usr/bin/env python3 """一个只依赖 OpenAI SDK 的教学版 Mini Agent。核心流程：用户问题 -> 调用 LLM -> 执行工具 -> 回填工具结果 -> 再次调用 LLM -> 最终答案。 """ from __future__ import annotations import argparse import json import logging import os from datetime import datetime from typing import Any, Callable from ddgs import DDGS from openai import OpenAI SYSTEM_PROMPT = ( "You are a helpful assistant. Use tools when they can improve accuracy. " "After receiving tool results, answer the user's question clearly." ) # 模型看到的是 JSON Schema，而不是下面的 Python 函数本身。 TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "get_current_time", "description": "Return the current local date and time.", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "calculate", "description": "Calculate a basic arithmetic expression.", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "Arithmetic expression, for example: (15 + 3) * 2", } }, "required": ["expression"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "web_search", "description": "Search the web for current or factual information.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Search keywords"} }, "required": ["query"], "additionalProperties": False, }, }, }, ] def get_current_time() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def calculate(expression: str) -> str: # 为了突出 Agent 主循环，这里沿用参考项目的极简实现；不要用于生产环境。 return str(eval(expression, {"__builtins__": {}}, {})) def web_search(query: str) -> str: """执行真实网页搜索，并返回标题、链接和摘要。""" results = list(DDGS().text(query, max_results=5)) return json.dumps(results, ensure_ascii=False, indent=2) if results else "No search results found." # 工具名是模型输出与本地 Python 函数之间的桥梁。 TOOL_FUNCTIONS: dict[str, Callable[..., str]] = { "get_current_time": get_current_time, "calculate": calculate, "web_search": web_search, } def execute_tool(name: str, arguments_json: str) -> str: """解析模型生成的参数并执行对应工具；错误也作为观察结果返回模型。""" try: arguments = json.loads(arguments_json) if not isinstance(arguments, dict): raise ValueError("工具参数必须是 JSON object") function = TOOL_FUNCTIONS.get(name) if function is None: raise ValueError(f"未知工具: {name}") return str(function(**arguments)) except Exception as exc: return f"Tool error: {type(exc).__name__}: {exc}" def _response_dict(response: Any) -> dict[str, Any]: """把官方 SDK 对象或兼容接口返回的 JSON 字符串统一转换为字典。""" if hasattr(response, "model_dump"): return response.model_dump() if isinstance(response, dict): return response if isinstance(response, str): try: data = json.loads(response) except json.JSONDecodeError as exc: if response.lstrip().lower().startswith((" dict[str, Any]: """只保存下一次 API 请求需要的 assistant 字段。""" result: dict[str, Any] = { "role": "assistant", "content": message.get("content"), } if message.get("tool_calls"): result["tool_calls"] = message["tool_calls"] return result def run_agent(question: str, client: OpenAI, model: str, max_steps: int = 8) -> str: """运行最小 Agent 循环，直到模型返回普通文本或达到最大步数。""" messages: list[dict[str, Any]] = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}, ] for step in range(1, max_steps + 1): logging.info("========== Agent step %d/%d ==========", step, max_steps) logging.info("OpenAI request messages:\n%s", json.dumps(messages, ensure_ascii=False, indent=2)) response = client.chat.completions.create( model=model, messages=messages, tools=TOOL_SCHEMAS, tool_choice="auto", ) # 官方 SDK 返回对象，部分兼容接口则可能返回 JSON 字符串，统一转为字典。 payload = _response_dict(response) logging.info( "OpenAI raw response:\n%s", json.dumps(payload, ensure_ascii=False, indent=2), ) try: message = payload["choices"][0]["message"] except (KeyError, IndexError, TypeError) as exc: raise RuntimeError("接口响应缺少 choices[0].message，可能不兼容 OpenAI 格式") from exc messages.append(_assistant_message(message)) tool_calls = message.get("tool_calls") or [] if not tool_calls: answer = message.get("content") or "" logging.info("Agent finished without another tool call.") return answer for tool_call in tool_calls: name = tool_call["function"]["name"] arguments = tool_call["function"]["arguments"] logging.info("Tool call: name=%s arguments=%s", name, arguments) result = execute_tool(name, arguments) logging.info("Tool result: name=%s result=%s", name, result) # tool_call_id 让模型知道该结果对应哪一次工具请求。 messages.append( { "role": "tool", "tool_call_id": tool_call["id"], "content": result, } ) raise RuntimeError(f"Agent 在 {max_steps} 步内没有完成任务") def main() -> None: parser = argparse.ArgumentParser(description="Minimal educational OpenAI agent") parser.add_argument("question", nargs="+", help="本次运行要回答的问题") parser.add_argument("--model", default=os.getenv("OPENAI_MODEL", "gpt-4o-mini")) parser.add_argument("--max-steps", type=int, default=8) parser.add_argument("--quiet", action="store_true", help="关闭过程日志，只输出答案") args = parser.parse_args() logging.basicConfig( level=logging.WARNING if args.quiet else logging.INFO, format=( "%(asctime)s | %(levelname)s | %(filename)s:%(lineno)d " "| %(funcName)s() | %(message)s" ), ) if not os.getenv("OPENAI_API_KEY"): parser.error("请先设置环境变量 OPENAI_API_KEY") if args.max_steps < 1: parser.error("--max-steps 必须大于 0") client = OpenAI( api_key=os.environ["OPENAI_API_KEY"], base_url=os.getenv("OPENAI_BASE_URL") or None, ) answer = run_agent(" ".join(args.question), client, args.model, args.max_steps) print(f"\nAnswer: {answer}") if __name__ == "__main__": main()