finai/backend/app/integrations/adp_client.py

600 lines
23 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import contextlib
import json
import os
import re
import time
import uuid
from collections.abc import Iterator
from pathlib import Path
import httpx
from app.config import get_settings
_BOT_BIZ_ID_CACHE: dict[str, str] = {}
@contextlib.contextmanager
def _without_proxy():
"""SDK 走 requests, 会读 HTTP_PROXY/HTTPS_PROXY. 本地代理抽风会 RESET 出站, 这里临时屏蔽."""
keys = ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY", "all_proxy")
saved = {k: os.environ.pop(k, None) for k in keys}
os.environ["NO_PROXY"] = "*"
try:
yield
finally:
os.environ.pop("NO_PROXY", None)
for k, v in saved.items():
if v is not None:
os.environ[k] = v
def _retry_sdk(call, *, attempts: int = 3, backoff: float = 0.6):
"""对 SDK 偶发 ClientNetworkError 自动重试."""
last_exc = None
for i in range(attempts):
try:
return call()
except Exception as exc:
msg = str(exc)
if "ClientNetworkError" not in msg and "Connection" not in msg and "timeout" not in msg.lower():
raise
last_exc = exc
if i + 1 < attempts:
time.sleep(backoff * (2**i))
raise last_exc
class AdpClient:
def __init__(self) -> None:
self._settings = get_settings()
def _build_client(self, timeout: int = 30):
s = self._settings
if not s.tencent_secret_id or not s.tencent_secret_key:
raise RuntimeError("缺少 TENCENT_SECRET_ID / TENCENT_SECRET_KEY请到腾讯云控制台-访问管理-访问密钥获取后填入 .env")
if not s.adp_bot_app_key:
raise RuntimeError("缺少 ADP_BOT_APP_KEY")
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.lke.v20231130 import lke_client, models
cred = credential.Credential(s.tencent_secret_id, s.tencent_secret_key)
http_profile = HttpProfile(reqMethod="POST", reqTimeout=timeout)
profile = ClientProfile(httpProfile=http_profile)
client = lke_client.LkeClient(cred, s.tencent_region, profile)
return client, models
def _resolve_bot_biz_id(self, client, models) -> str:
s = self._settings
bot_biz_id = _BOT_BIZ_ID_CACHE.get(s.adp_bot_app_key)
if bot_biz_id:
return bot_biz_id
bot_req = models.DescribeRobotBizIDByAppKeyRequest()
bot_req.AppKey = s.adp_bot_app_key
bot_resp = _retry_sdk(lambda: client.DescribeRobotBizIDByAppKey(bot_req))
bot_biz_id = bot_resp.BotBizId
if not bot_biz_id:
raise RuntimeError("DescribeRobotBizIDByAppKey 没返回 BotBizId")
_BOT_BIZ_ID_CACHE[s.adp_bot_app_key] = bot_biz_id
return bot_biz_id
def _resolve_kb_biz_id(self, client, models) -> str:
"""对文档管理 APIListDoc/SaveDoc/DeleteDoc/ModifyDoc使用的 BotBizId。
优先使用配置里的共享知识库 ID没配则回退到应用自身的 robot biz id。
"""
configured = (self._settings.adp_knowledge_biz_id or "").strip()
if configured:
return configured
return self._resolve_bot_biz_id(client, models)
def list_documents(
self,
page: int = 1,
page_size: int = 50,
query: str = "",
cate_biz_id: str = "",
) -> dict:
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
list_req = models.ListDocRequest()
list_req.BotBizId = bot_biz_id
list_req.PageNumber = page
list_req.PageSize = page_size
if query:
list_req.Query = query
if cate_biz_id:
list_req.CateBizId = cate_biz_id
list_resp = _retry_sdk(lambda: client.ListDoc(list_req))
return {
"total": int(list_resp.Total or 0),
"list": [_serialize_doc(item) for item in (list_resp.List or [])],
}
def list_categories(self) -> list[dict]:
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
req = models.ListDocCateRequest()
req.BotBizId = bot_biz_id
resp = _retry_sdk(lambda: client.ListDocCate(req))
return [_serialize_category(item) for item in (resp.List or [])]
def create_category(self, name: str, parent_biz_id: str = "") -> dict:
if not name:
raise RuntimeError("分类名称不能为空")
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
req = models.CreateDocCateRequest()
req.BotBizId = bot_biz_id
req.Name = name
if parent_biz_id:
req.ParentBizId = parent_biz_id
resp = _retry_sdk(lambda: client.CreateDocCate(req))
return {"cate_biz_id": getattr(resp, "CateBizId", "") or ""}
def modify_category(self, cate_biz_id: str, name: str) -> None:
if not cate_biz_id:
raise RuntimeError("缺少 CateBizId")
if not name:
raise RuntimeError("分类名称不能为空")
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
req = models.ModifyDocCateRequest()
req.BotBizId = bot_biz_id
req.CateBizId = cate_biz_id
req.Name = name
_retry_sdk(lambda: client.ModifyDocCate(req))
def delete_category(self, cate_biz_id: str) -> None:
if not cate_biz_id:
raise RuntimeError("缺少 CateBizId")
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
req = models.DeleteDocCateRequest()
req.BotBizId = bot_biz_id
req.CateBizId = cate_biz_id
_retry_sdk(lambda: client.DeleteDocCate(req))
def delete_documents(self, doc_biz_ids: list[str]) -> None:
ids = [i for i in doc_biz_ids if i]
if not ids:
raise RuntimeError("缺少 DocBizId")
with _without_proxy():
client, models = self._build_client(timeout=30)
bot_biz_id = self._resolve_kb_biz_id(client, models)
req = models.DeleteDocRequest()
req.BotBizId = bot_biz_id
req.DocBizIds = ids
_retry_sdk(lambda: client.DeleteDoc(req))
def upload_chat_image(self, file_bytes: bytes, *, mime_type: str, original_name: str = "") -> str:
"""上传聊天图片到 ADP 实时 COS, 返回 /adp/v2/chat ImageUrl 可直接使用的公网 HTTPS URL.
与 upload_document 的离线知识库上传不同, 这里走 TypeKey=realtime + IsPublic=True.
"""
file_type = _image_ext_from_mime(mime_type, original_name)
if not file_type:
raise RuntimeError("图片格式必须是 png/jpg/jpeg")
with _without_proxy():
client, models = self._build_client(timeout=60)
bot_biz_id = self._resolve_kb_biz_id(client, models)
cred_req = models.DescribeStorageCredentialRequest()
cred_req.BotBizId = bot_biz_id
cred_req.FileType = file_type
cred_req.IsPublic = True
cred_req.TypeKey = "realtime"
cred_resp = _retry_sdk(lambda: client.DescribeStorageCredential(cred_req))
upload_url = getattr(cred_resp, "UploadUrl", "") or ""
upload_path = getattr(cred_resp, "UploadPath", "") or ""
if not upload_url or not upload_path:
raise RuntimeError("DescribeStorageCredential 未返回 UploadUrl/UploadPath")
with httpx.Client(timeout=120, trust_env=True) as http:
put_resp = http.put(
upload_url,
content=file_bytes,
headers={"Content-Type": mime_type or "application/octet-stream"},
)
put_resp.raise_for_status()
file_url = getattr(cred_resp, "FileUrl", "") or ""
if file_url:
return file_url
bucket = getattr(cred_resp, "Bucket", "") or ""
region = getattr(cred_resp, "Region", "") or ""
if bucket and region:
return f"https://{bucket}.cos.{region}.myqcloud.com{upload_path}"
raise RuntimeError("DescribeStorageCredential 未返回 FileUrl/Bucket/Region")
def upload_document(self, file_name: str, file_bytes: bytes) -> dict:
file_type = Path(file_name).suffix.lstrip(".").lower()
if not file_type:
raise RuntimeError("无法识别文件类型,请确保文件名含扩展名")
with _without_proxy():
client, models = self._build_client(timeout=60)
bot_biz_id = self._resolve_kb_biz_id(client, models)
cred_req = models.DescribeStorageCredentialRequest()
cred_req.BotBizId = bot_biz_id
cred_req.FileType = file_type
cred_req.IsPublic = False
cred_req.TypeKey = "offline"
cred_resp = _retry_sdk(lambda: client.DescribeStorageCredential(cred_req))
upload_url = cred_resp.UploadUrl
upload_path = cred_resp.UploadPath
if not upload_url or not upload_path:
raise RuntimeError("DescribeStorageCredential 未返回 UploadUrl/UploadPath")
with httpx.Client(timeout=120, trust_env=True) as http:
put_resp = http.put(
upload_url,
content=file_bytes,
headers={"Content-Type": "application/octet-stream"},
)
put_resp.raise_for_status()
etag = (put_resp.headers.get("ETag") or "").strip('"')
cos_hash = put_resp.headers.get("x-cos-hash-crc64ecma") or ""
if not etag or not cos_hash:
raise RuntimeError("PUT 上传后未拿到 ETag 或 x-cos-hash-crc64ecma")
save_req = models.SaveDocRequest()
save_req.BotBizId = bot_biz_id
save_req.FileName = file_name
save_req.FileType = file_type
save_req.CosUrl = upload_path
save_req.ETag = etag
save_req.CosHash = cos_hash
save_req.Size = str(len(file_bytes))
save_req.Source = 0
save_req.EnableScope = 4
save_resp = _retry_sdk(lambda: client.SaveDoc(save_req))
return {
"doc_biz_id": save_resp.DocBizId or "",
"file_name": file_name,
"error_msg": save_resp.ErrorMsg or "",
"duplicate_check_type": save_resp.DuplicateFileCheckType or 0,
}
def stream_answer(
self,
question: str,
*,
user_id: str | None = None,
conversation_id: str | None = None,
allowed_tags: list[str] | None = None,
image_urls: list[str] | None = None,
) -> Iterator[dict]:
if self._settings.adp_provider == "tencent":
yield from self._stream_visitor(
question,
user_id=user_id or "anonymous",
conversation_id=conversation_id or uuid.uuid4().hex,
allowed_tags=allowed_tags or [],
image_urls=image_urls or [],
)
else:
yield from self._stream_mock()
def _stream_mock(self) -> Iterator[dict]:
answer = (
"根据当前知识库,建议先确认制度适用范围,再按材料准备、系统填报、审批提交、财务复核的顺序办理。"
"关键事项仍以正式制度文件和业务部门解释为准。"
)
for chunk in [answer[:32], answer[32:64], answer[64:]]:
if chunk:
yield {"event": "answer.delta", "data": chunk}
yield {
"event": "answer.reference",
"data": {
"source_title": "《差旅费报销管理办法》",
"snippet": "报销材料、审批单、交通与住宿票据要求。",
"page_no": "3",
"section": "报销材料",
"source_url": None,
},
}
yield {"event": "answer.usage", "data": {"token_usage": 128, "pu_usage": 1}}
yield {"event": "answer.done", "data": {}}
def _stream_visitor(
self,
question: str,
*,
user_id: str,
conversation_id: str,
allowed_tags: list[str],
image_urls: list[str] | None = None,
) -> Iterator[dict]:
s = self._settings
# ADP 对话接口不支持独立的 image 子项,图片以 markdown 链接 ![](url)
# 的形式嵌在文本里传入。参考: https://cloud.tencent.com/document/product/1759/105561
text = (question or "").strip()
for url in (image_urls or []):
text = f"{text}\n![]({url})" if text else f"![]({url})"
if not text:
text = "请回答这张图片相关的问题。"
contents: list[dict] = [{"Type": "text", "Text": text}]
if allowed_tags:
contents.append({"Type": "custom_variables", "CustomVariables": {"allowed_tags": ",".join(allowed_tags)}})
body = {
"RequestId": uuid.uuid4().hex,
"ConversationId": conversation_id,
"AppKey": s.adp_bot_app_key,
"VisitorId": user_id,
"Contents": contents,
"Stream": "enable",
"Incremental": True,
"EnableMultiIntent": True,
}
headers = {"Content-Type": "application/json"}
answer_buf: list[str] = []
with httpx.Client(timeout=s.adp_timeout_seconds) as client:
with client.stream(
"POST",
s.adp_endpoint,
headers=headers,
content=json.dumps(body, ensure_ascii=False).encode("utf-8"),
) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
line = (line or "").strip()
if not line or not line.startswith("data:"):
continue
data_str = line[5:].strip()
if data_str in {"", "[DONE]"}:
continue
try:
event = json.loads(data_str)
except json.JSONDecodeError:
continue
print("[ADP EVENT]", json.dumps(event, ensure_ascii=False), flush=True) # DEBUG remove after
t = (event.get("Type") or "").lower()
if t in {"text.delta", "text.replace"}:
answer_buf.append(event.get("Text") or "")
yield from _translate_adp_event(event, "".join(answer_buf))
yield {"event": "answer.done", "data": {}}
_IMAGE_MIME_EXT = {
"image/png": "png",
"image/jpeg": "jpg",
"image/jpg": "jpg",
}
def _image_ext_from_mime(mime_type: str, original_name: str = "") -> str:
ext = _IMAGE_MIME_EXT.get((mime_type or "").lower())
if ext:
return ext
suffix = Path(original_name or "").suffix.lstrip(".").lower()
if suffix == "jpeg":
return "jpg"
if suffix in {"png", "jpg"}:
return suffix
return ""
_SNIPPET_MAX = 500
def _parse_knowledge_content(content: str) -> tuple[str, str]:
"""ADP Knowledge.Outputs[].Content 形如:
文档名:差旅费(母公司)\n文档片段:\n第八条 员工出差...
返回 (source_title, full_body);调用方负责再聚焦/截断。
"""
title = ""
body = content
if content.startswith("文档名:"):
head, _, rest = content.partition("\n")
title = head.removeprefix("文档名:").strip()
if rest.startswith("文档片段:") or rest.startswith("文档片段:\n"):
body = rest.removeprefix("文档片段:").lstrip("\n").strip()
else:
body = rest
return title or "未知来源", body
def _focus_snippet(chunk: str, answer: str, max_chars: int = _SNIPPET_MAX) -> str:
"""从 chunk 里挑出被 answer 真正引用的那段。
优先用答案前缀做 needle, 找不到再用答案里的各短语找命中段落, 都失败回退到 chunk 开头.
"""
if not chunk:
return ""
if not answer:
return chunk[:max_chars]
needle = answer.strip().rstrip("。.;!?")
idx = -1
for size in (40, 30, 22, 16, 10):
if len(needle) < size:
continue
idx = chunk.find(needle[:size])
if idx >= 0:
break
# 前缀没命中, 再用答案各短语找一次最早出现的命中点
if idx < 0:
fragments = [
f.strip() for f in re.split(r"[。.;!?\n ,、:()【】\[\]]", answer)
if len(f.strip()) >= 8
]
for frag in fragments:
j = chunk.find(frag)
if j >= 0 and (idx < 0 or j < idx):
idx = j
if idx < 0:
return chunk[:max_chars]
para_start = chunk.rfind("\n\n", 0, idx)
para_start = 0 if para_start < 0 else para_start + 2
para_end = chunk.find("\n\n", idx)
para_end = len(chunk) if para_end < 0 else para_end
snippet = chunk[para_start:para_end].strip()
if len(snippet) > max_chars:
rel = idx - para_start
half = max_chars // 2
s = max(0, rel - half)
e = min(len(snippet), s + max_chars)
snippet = snippet[s:e]
if s > 0:
snippet = "" + snippet
if e < (para_end - para_start):
snippet = snippet + ""
return snippet
def _translate_adp_event(event: dict, answer_text: str = "") -> Iterator[dict]:
event_type = (event.get("Type") or "").lower()
if event_type in {"text.delta", "text.replace"}:
text = event.get("Text") or ""
if text:
yield {"event": "answer.delta", "data": text}
elif event_type == "response.completed":
response = event.get("Response") or {}
stat = response.get("StatInfo") or {}
if stat:
yield {
"event": "answer.usage",
"data": {
"token_usage": int(stat.get("TotalTokens") or 0),
"pu_usage": int(stat.get("TotalCost") or 0),
},
}
# ADP 把"召回的所有候选"放在 Procedures[].Knowledge.Outputs (含正文片段),
# 把"LLM 实际引用的文档"放在 Messages[].Contents[].References (只有标题, 无片段).
# 有 References 时以 References 为权威列表, 按标题回填 snippet;
# 没 References 时 (澄清反问/LLM 未引用) 直接把召回候选按顺序全量输出为"参考知识片段", 与线上行为对齐.
all_chunks: list[tuple[str, str]] = []
for procedure in response.get("Procedures") or []:
if (procedure.get("Type") or "").lower() != "knowledge":
continue
for output in (procedure.get("Knowledge") or {}).get("Outputs") or []:
content = output.get("Content") or ""
if not content:
continue
all_chunks.append(_parse_knowledge_content(content))
snippets_by_title: dict[str, str] = {}
for title, body in all_chunks:
snippets_by_title.setdefault(title, _focus_snippet(body, answer_text))
yielded_any = False
for message in response.get("Messages") or []:
for content_item in message.get("Contents") or []:
for ref in content_item.get("References") or []:
doc_ref = ref.get("DocRefer") or {}
title = (ref.get("Name") or doc_ref.get("DocName") or "").strip()
if not title:
continue
yielded_any = True
yield {
"event": "answer.reference",
"data": {
"source_title": title,
"snippet": snippets_by_title.get(title, ""),
"page_no": "",
"section": None,
"source_url": doc_ref.get("Url") or None,
},
}
if not yielded_any and all_chunks:
for title, body in all_chunks:
yield {
"event": "answer.reference",
"data": {
"source_title": title,
"snippet": _focus_snippet(body, answer_text),
"page_no": "",
"section": None,
"source_url": None,
},
}
elif event_type == "error":
err = event.get("Error") or {}
yield {
"event": "answer.delta",
"data": f"\n[ADP 错误 {err.get('Code')}] {err.get('Message') or '请稍后重试'}",
}
def _serialize_category(item) -> dict:
def _opt(name):
return getattr(item, name, None)
try:
total = int(_opt("Total") or 0)
except (TypeError, ValueError):
total = 0
children_raw = _opt("Children") or []
return {
"cate_biz_id": _opt("CateBizId") or "",
"name": _opt("Name") or "",
"total": total,
"can_add": bool(_opt("CanAdd")),
"can_edit": bool(_opt("CanEdit")),
"can_delete": bool(_opt("CanDelete")),
"is_leaf": bool(_opt("IsLeaf")),
"children": [_serialize_category(c) for c in children_raw],
}
def _serialize_doc(item) -> dict:
def _opt(name):
return getattr(item, name, None)
size_bytes = _opt("DocSize")
try:
size_bytes = int(size_bytes) if size_bytes else 0
except (TypeError, ValueError):
size_bytes = 0
attr_labels: list[dict] = []
for attr in (_opt("AttrLabels") or []):
labels = []
for label in (getattr(attr, "Labels", None) or []):
labels.append({
"label_biz_id": getattr(label, "LabelBizId", "") or "",
"label_name": getattr(label, "LabelName", "") or "",
})
attr_labels.append({
"attr_biz_id": getattr(attr, "AttrBizId", "") or "",
"attr_name": getattr(attr, "AttrName", "") or "",
"attr_key": getattr(attr, "AttrKey", "") or "",
"labels": labels,
})
return {
"doc_biz_id": _opt("DocBizId") or "",
"file_name": _opt("FileName") or "",
"file_type": _opt("FileType") or "",
"size_bytes": size_bytes,
"status": _opt("Status") or 0,
"status_desc": _opt("StatusDesc") or "",
"source": _opt("Source") or 0,
"source_desc": _opt("SourceDesc") or "",
"is_disabled": bool(_opt("IsDisabled")),
"enable_scope": _opt("EnableScope") or 0,
"update_time": _opt("UpdateTime") or "",
"create_time": _opt("CreateTime") or "",
"char_size": _opt("DocCharSize") or "",
"qa_num": _opt("QaNum") or 0,
"category_biz_id": _opt("CateBizId") or "",
"attr_labels": attr_labels,
}