吉鑫升andClaude Fable 5
28098b3155
- 知识库笔记列表支持按实体精确筛选（路由/服务/仓储层透传，关系图索引查询） - 自定义智能体技能白名单在 skill_view 与 skill_manage 的 list/view 别名上同样生效； config 未声明 skills 时视为无可见技能 - 嵌入圆桌主题解析默认深蓝，强制主题期间不再写 localStorage 覆盖主站选择 - 智能体提问建议拆出 agent-question-chips 组件，编辑器与详情面板联动优化 - 新增 qa-client 问答测试脚本 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
136 lines
5.4 KiB
Python
136 lines
5.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""DeerFlow 后端问答客户端（零第三方依赖，纯 Python 标准库）。
|
||
|
||
用法：只需修改下面的 BASE_URL（必要时改 USERNAME / LOGIN_PASSWORD），
|
||
然后调用 ask("你的问题") 即可拿到回答文本。
|
||
|
||
from qa_client import ask
|
||
|
||
answer = ask("你好，介绍一下你自己")
|
||
print(answer)
|
||
|
||
多轮对话（共用同一个 thread_id 即可保留上下文）：
|
||
|
||
tid = new_thread_id()
|
||
print(ask("我叫小明", thread_id=tid))
|
||
print(ask("我叫什么名字？", thread_id=tid))
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import ssl
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
# ======================= 只需要修改这里 =======================
|
||
BASE_URL = "http://47.94.209.59:2026" # 后端 Gateway 地址（改成你的服务器地址即可）
|
||
USERNAME = "api-client" # 登录用户名（免密登录，不存在会自动注册）
|
||
LOGIN_PASSWORD = "123ewq" # 仅当后端开启 require_password 口令门槛时才需要填（默认口令 123ewq）
|
||
MODEL_NAME = None # 可选：指定模型名；None = 使用后端默认模型
|
||
TIMEOUT_SECONDS = 600 # 等待回答的超时时间（复杂问题可能跑几分钟）
|
||
# =============================================================
|
||
|
||
_token_cache = {"token": None}
|
||
_ssl_ctx = ssl.create_default_context()
|
||
_ssl_ctx.check_hostname = False
|
||
_ssl_ctx.verify_mode = ssl.CERT_NONE # 内网自签证书也能用
|
||
|
||
|
||
def _post_json(url: str, payload: dict, token: str | None = None, timeout: int = 30) -> dict:
|
||
"""POST JSON 并解析 JSON 响应。"""
|
||
data = json.dumps(payload).encode("utf-8")
|
||
req = urllib.request.Request(url, data=data, method="POST")
|
||
req.add_header("Content-Type", "application/json")
|
||
if token:
|
||
req.add_header("Authorization", f"Bearer {token}")
|
||
with urllib.request.urlopen(req, timeout=timeout, context=_ssl_ctx) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def _login() -> str | None:
|
||
"""免密用户名登录，返回 Bearer token；后端关闭鉴权时返回 None 也能正常问答。"""
|
||
if _token_cache["token"]:
|
||
return _token_cache["token"]
|
||
url = f"{BASE_URL.rstrip('/')}/api/v1/auth/login/username"
|
||
if LOGIN_PASSWORD:
|
||
url += "?" + urllib.parse.urlencode({"password": LOGIN_PASSWORD})
|
||
try:
|
||
result = _post_json(url, {"username": USERNAME})
|
||
_token_cache["token"] = result.get("access_token")
|
||
return _token_cache["token"]
|
||
except urllib.error.HTTPError as e:
|
||
if e.code == 401:
|
||
raise RuntimeError("登录被拒绝：后端开启了登录口令，请在 LOGIN_PASSWORD 填入正确口令（默认 123ewq）") from e
|
||
# 登录接口异常时尝试无 token 直连（后端可能处于免鉴权模式）
|
||
return None
|
||
except urllib.error.URLError:
|
||
return None
|
||
|
||
|
||
def _extract_answer(channel_values: dict) -> str:
|
||
"""从 /api/runs/wait 返回的最终状态里取出最后一条 AI 回复文本。"""
|
||
messages = channel_values.get("messages") or []
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict):
|
||
continue
|
||
if msg.get("type") not in ("ai", "AIMessageChunk", "AIMessage"):
|
||
continue
|
||
content = msg.get("content")
|
||
if isinstance(content, str) and content.strip():
|
||
return content.strip()
|
||
if isinstance(content, list): # 结构化内容块：拼接其中的文本
|
||
parts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
|
||
text = "".join(parts).strip()
|
||
if text:
|
||
return text
|
||
# 没有 AI 消息时，把原始返回吐出来便于排查
|
||
return json.dumps(channel_values, ensure_ascii=False)[:2000]
|
||
|
||
|
||
def new_thread_id() -> str:
|
||
"""生成一个新会话 id，用于多轮对话。"""
|
||
return str(uuid.uuid4())
|
||
|
||
|
||
def ask(question: str, thread_id: str | None = None) -> str:
|
||
"""向后端发送一个问题，阻塞直到拿到最终回答文本。
|
||
|
||
:param question: 要问的问题
|
||
:param thread_id: 可选。传同一个 id 即为多轮对话（保留上下文）；不传则每次新会话
|
||
:return: AI 的回答文本
|
||
"""
|
||
token = _login()
|
||
configurable: dict = {}
|
||
if thread_id:
|
||
configurable["thread_id"] = thread_id
|
||
if MODEL_NAME:
|
||
configurable["model_name"] = MODEL_NAME
|
||
|
||
payload = {
|
||
"input": {"messages": [{"role": "user", "content": question}]},
|
||
"config": {"configurable": configurable},
|
||
}
|
||
url = f"{BASE_URL.rstrip('/')}/api/runs/wait"
|
||
try:
|
||
result = _post_json(url, payload, token=token, timeout=TIMEOUT_SECONDS)
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode("utf-8", errors="replace")[:500]
|
||
raise RuntimeError(f"请求失败 HTTP {e.code}: {body}") from e
|
||
except urllib.error.URLError as e:
|
||
raise RuntimeError(f"无法连接后端 {BASE_URL}，请检查 BASE_URL 和服务是否启动: {e.reason}") from e
|
||
|
||
return _extract_answer(result)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 示例 1：单轮问答
|
||
print(ask("你好，请简单介绍一下你自己"))
|
||
|
||
# 示例 2：多轮对话（取消注释即可）
|
||
# tid = new_thread_id()
|
||
# print(ask("我叫小明，记住了吗？", thread_id=tid))
|
||
# print(ask("我叫什么名字？", thread_id=tid))
|