1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126
| """最小 ReAct Agent:模型可以调用本地计算器,而不是自己心算。
判断某一轮走了哪条路,看这两个信号: - finish_reason == "tool_calls" 且 msg.tool_calls 有值 -> 跑了本地工具 - finish_reason == "stop" 且 msg.tool_calls 为空 -> 给出最终自然语言回答 """
import json import os from pathlib import Path
from openai import OpenAI
OPENCODE_GO_BASE_URL = "https://opencode.ai/zen/go/v1" MODEL = os.environ.get("OPENCODE_MODEL", "glm-5.1")
def load_api_key() -> str: """优先读环境变量,没有再用本机 OpenCode 登录态里的 key。""" key = os.environ.get("OPENCODE_API_KEY") or os.environ.get("OPENCODE_GO_API_KEY") if key: return key auth_path = Path.home() / ".local/share/opencode/auth.json" if auth_path.exists(): data = json.loads(auth_path.read_text()) for provider in ("opencode-go", "opencode"): stored = data.get(provider) or {} if stored.get("key"): return stored["key"]
raise RuntimeError( "Missing OpenCode API key. Set OPENCODE_API_KEY, or log in with `opencode`." )
client = OpenAI(base_url=OPENCODE_GO_BASE_URL, api_key=load_api_key())
def calculate(expression: str) -> str: try: return str(eval(expression)) except Exception as e: return f"Error: {e}"
tool_map = {"calculate": calculate}
tools = [{ "type": "function", "function": { "name": "calculate", "description": "Evaluate a math expression", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "For example: 123 * 456", } }, "required": ["expression"], }, }, }]
messages = [{"role": "user", "content": "帮我算一下 248 乘以 15 等于多少?"}]
round_num = 0 while True: round_num += 1 response = client.chat.completions.create( model=MODEL, messages=messages, tools=tools, ) choice = response.choices[0] msg = choice.message print(f"\n--- round {round_num} ---") print("finish_reason:", choice.finish_reason)
assistant_msg = {"role": "assistant", "content": msg.content or ""} if msg.tool_calls: assistant_msg["tool_calls"] = [ { "id": tool_call.id, "type": "function", "function": { "name": tool_call.function.name, "arguments": tool_call.function.arguments, }, } for tool_call in msg.tool_calls ] messages.append(assistant_msg)
if not msg.tool_calls: print("path: model returned the answer directly (no tool_calls)") print("Agent final answer:", msg.content) break
print("path: model requested tool_calls") for tool_call in msg.tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments)
result = tool_map[func_name](**func_args) print(f" call {func_name}({func_args}) -> {result}") messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result, })
|