sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
r"""真实 API 测试：opencode-go / deepseek-v4-flash + 完整 agent 循环（工具执行）
|
|
运行: C:\Users\14890\miniconda3\envs\haocode\python.exe -u tests/diag_live_agent.py
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from core.agent import (Agent, AgentConfig, ModelConfig, RetryConfig) # noqa: E402
|
|
from core.agent.recovery import AgentRunner # noqa: E402
|
|
from core.agent.stream_fn import openai_stream # noqa: E402
|
|
from core.agent.tools import default_tools # noqa: E402
|
|
|
|
MODEL_NAME = "deepseek-v4-flash"
|
|
|
|
cfg = json.load(open(os.path.join(ROOT, "data", "config.json"), encoding="utf-8"))
|
|
prov = cfg["providers"]["opencode-go"]
|
|
|
|
model = ModelConfig(
|
|
provider="opencode-go", name=MODEL_NAME,
|
|
context_window=int(prov.get("model_contexts", {}).get(MODEL_NAME, 1000000)),
|
|
max_tokens=4096, temperature=0.3,
|
|
api_key=prov["api_key"], base_url=prov["base_url"],
|
|
)
|
|
|
|
with open(os.path.join(ROOT, "SYSTEM_PROMPT.md"), encoding="utf-8") as f:
|
|
system_prompt = f.read()
|
|
|
|
agent_cfg = AgentConfig(
|
|
model=model,
|
|
tools=default_tools(),
|
|
system_prompt=system_prompt,
|
|
tool_context={"cwd": ROOT},
|
|
retry=RetryConfig(max_attempts=2, base_delay_ms=1000),
|
|
)
|
|
agent = Agent(agent_cfg)
|
|
agent.set_stream_fn(openai_stream)
|
|
|
|
|
|
def show(e):
|
|
if e.type == "message_update" and e.assistant_message_event is not None:
|
|
ev = e.assistant_message_event
|
|
if ev.type == "text_delta":
|
|
print(" [正文] " + ev.text, end="", flush=True)
|
|
elif ev.type == "thinking_delta":
|
|
print(" [思考] " + ev.text, end="", flush=True)
|
|
elif ev.type == "toolcall_delta":
|
|
print(f" [tc] {ev.tool_call_field}={ev.tool_call_delta!r}",
|
|
end="", flush=True)
|
|
elif e.type == "tool_execution_start" and e.tool_call is not None:
|
|
print(f"\n [工具开始] {e.tool_call.name} 参数={e.tool_call.arguments}")
|
|
elif e.type == "tool_execution_update" and e.arg:
|
|
print(" [工具输出] " + str(e.arg), end="", flush=True)
|
|
elif e.type == "tool_execution_end" and e.tool_call is not None:
|
|
c = e.result.content if e.result else ""
|
|
if not isinstance(c, str):
|
|
c = "".join(x.get("text", "") for x in c if isinstance(x, dict))
|
|
print(f"\n [工具结束] ok={not e.is_error} 结果={c[:200]!r}")
|
|
elif e.type == "agent_end":
|
|
print(f"\n [agent_end] stop_reason={e.stop_reason} "
|
|
f"error={getattr(e.error, 'message', None)}")
|
|
|
|
|
|
agent.subscribe(show)
|
|
runner = AgentRunner(agent)
|
|
|
|
print("=" * 60)
|
|
print(f"实时测试: {model.base_url} / {MODEL_NAME}")
|
|
print(f"system prompt: {len(system_prompt)} 字符, tools: {len(default_tools())} 个")
|
|
print("=" * 60)
|
|
|
|
runner.run("请用 bash 工具执行命令: echo hello-from-haocode && ls，然后告诉我输出结果。")
|
|
|
|
print("\n===== 最终消息链 =====")
|
|
for m in agent.state.messages:
|
|
tc = f" tool_calls={[t.name for t in m.tool_calls]}" if m.tool_calls else ""
|
|
print(f"- {m.role}: {(m.content or '')[:100]!r}{tc}")
|
|
|
|
ok = any(m.role == "toolResult" for m in agent.state.messages)
|
|
print("\n===== 结论:", "✅ 真实 tool_call 被发出并执行" if ok
|
|
else "❌ 没有工具执行（可能供应商不支持 tools API，检查是否走了文字兜底）", "=====")
|
|
sys.exit(0 if ok else 1)
|