78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""真实 DeepSeek → SearXNG → DeepSeek 联网闭环验收。
|
|
|
|
需要服务器密钥与已启动的 SearXNG，不属于日常离线回归测试。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.config import AppConfig # noqa: E402
|
|
from app.llm import create_provider # noqa: E402
|
|
from app.search import ( # noqa: E402
|
|
WEB_SEARCH_PROMPT,
|
|
WEB_SEARCH_TOOL,
|
|
SearXNGSearchClient,
|
|
)
|
|
|
|
|
|
async def main() -> int:
|
|
cfg = AppConfig.load()
|
|
if not cfg.search.enabled:
|
|
raise RuntimeError("请先设置 SEARCH_ENABLED=true")
|
|
|
|
llm = create_provider(cfg.llm)
|
|
search = SearXNGSearchClient(cfg.search)
|
|
messages = [
|
|
{"role": "system", "content": WEB_SEARCH_PROMPT},
|
|
{"role": "user", "content": "请搜索 DeepSeek 最近有什么消息，用一句话概括。"},
|
|
]
|
|
try:
|
|
first = await llm.complete(messages, tools=[WEB_SEARCH_TOOL])
|
|
if not first.tool_calls:
|
|
raise AssertionError(f"模型没有调用搜索工具，直接回答：{first.content!r}")
|
|
call = first.tool_calls[0]
|
|
query = call.parsed_arguments().get("query")
|
|
if not isinstance(query, str) or not query.strip():
|
|
raise AssertionError(f"搜索参数无效：{call.arguments!r}")
|
|
|
|
results = await search.search(query)
|
|
if not results:
|
|
raise AssertionError(f"SearXNG 对 {query!r} 没有返回结果")
|
|
messages.extend([
|
|
{
|
|
"role": "assistant",
|
|
"content": first.content or None,
|
|
"tool_calls": [{
|
|
"id": call.id,
|
|
"type": "function",
|
|
"function": {"name": call.name, "arguments": call.arguments},
|
|
}],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": call.id,
|
|
"content": search.for_model(query, results),
|
|
},
|
|
])
|
|
final = await llm.complete(messages, max_tokens=250)
|
|
if not final.content.strip():
|
|
raise AssertionError("搜索回灌后模型没有生成回答")
|
|
print(f"工具调用: {call.name}({query!r})")
|
|
print(f"搜索结果: {len(results)} 条；首条={results[0].title!r}")
|
|
print(f"最终回答: {final.content.strip()}")
|
|
print("真实联网闭环：PASS")
|
|
return 0
|
|
finally:
|
|
await llm.aclose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|