207 lines
8.4 KiB
Python
207 lines
8.4 KiB
Python
"""外部知识入库接口（发布 Markdown 产物到 knowes 知识库）。
|
|
|
|
智能体 config.yaml 配置了 ``template_name`` 后，该智能体会话中生成的 Markdown
|
|
产物可以通过 ``POST /api/knowledge-ingest/publish`` 一键入库。
|
|
|
|
入库目标接口的字段名不固定：标题字段可能叫「名称」也可能叫「标题」，因此先按
|
|
「名称」发送一次，若判定失败则用 **同一个 nodeName** 把标题字段换成「标题」再
|
|
发一次。nodeName 是入库记录的唯一标识，取「md 标题 + 时间戳」。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import hashlib
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from deerflow.config import get_app_config
|
|
from deerflow.config.agents_config import load_agent_config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/knowledge-ingest", tags=["knowledge-ingest"])
|
|
|
|
# 入库接口地址，可用环境变量覆盖（如切换 UAT/生产）。
|
|
_REQUEST_TIMEOUT_SECONDS = 30.0
|
|
|
|
# 标题字段在目标模板里可能叫「名称」或「标题」，按此顺序逐个尝试。
|
|
_TITLE_FIELD_CANDIDATES = ("名称", "标题")
|
|
_CONTENT_FIELD = "内容"
|
|
|
|
|
|
class KnowledgeIngestPublishRequest(BaseModel):
|
|
"""Request body for publishing one markdown artifact."""
|
|
|
|
agent_id: str = Field(..., description="智能体 id，用于读取其 config.yaml 中的 template_name")
|
|
title: str = Field(..., description="md 标题（一级标题或文件名）")
|
|
content: str = Field(..., description="md 正文全文")
|
|
|
|
|
|
class KnowledgeIngestAttempt(BaseModel):
|
|
"""One delivery attempt against the ingest API."""
|
|
|
|
title_field: str = Field(..., description="本次尝试使用的标题字段名（名称/标题）")
|
|
status_code: int | None = Field(default=None, description="HTTP 状态码；请求异常时为 null")
|
|
response_text: str = Field(default="", description="入库接口原始响应（截断到 2000 字符）")
|
|
success: bool = False
|
|
error: str | None = Field(default=None, description="请求层异常信息（网络/超时等）")
|
|
|
|
|
|
class KnowledgeIngestPublishResponse(BaseModel):
|
|
success: bool
|
|
template_name: str
|
|
node_name: str = Field(..., description="入库唯一标识（md 标题 + 时间戳），两次尝试共用")
|
|
batch_number: str
|
|
attempts: list[KnowledgeIngestAttempt]
|
|
|
|
|
|
def _judge_success(status_code: int, response_text: str) -> bool:
|
|
"""判断入库接口响应是否表示成功。
|
|
|
|
接口未提供正式契约：优先看响应 JSON 中常见的 success/code/status/state
|
|
字段；JSON 不可解析或没有这类字段时，退化为按 HTTP 2xx 判定。
|
|
"""
|
|
if not (200 <= status_code < 300):
|
|
return False
|
|
try:
|
|
data = json.loads(response_text)
|
|
except (ValueError, TypeError):
|
|
return True
|
|
if not isinstance(data, dict):
|
|
return True
|
|
if isinstance(data.get("success"), bool):
|
|
return data["success"]
|
|
for key in ("code", "status", "state"):
|
|
if key in data:
|
|
return str(data[key]).lower() in {"0", "200", "success", "ok", "true"}
|
|
return True
|
|
|
|
|
|
def _build_payload(template_name: str, batch_number: str, node_name: str, title_field: str, title: str, content: str) -> dict[str, Any]:
|
|
return {
|
|
"templateName": template_name,
|
|
"batchNumber": batch_number,
|
|
"list": [
|
|
{
|
|
"rec": {"nodeName": node_name},
|
|
"data": [
|
|
{"fieldName": title_field, "value": title},
|
|
{"fieldName": _CONTENT_FIELD, "value": content},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _build_node_name(title: str, timestamp: str) -> str:
|
|
"""Build a stable node name as md5(title) + timestamp + .md."""
|
|
title_md5 = hashlib.md5(title.encode("utf-8")).hexdigest()
|
|
return f"{title_md5}-{timestamp}.md"
|
|
|
|
|
|
def _resolve_ingest_url() -> str:
|
|
"""Resolve the external ingest URL from config.yaml, with env fallback."""
|
|
config_url = getattr(get_app_config().knowledge_ingest, "url", "").strip()
|
|
return os.environ.get(
|
|
"KNOWES_INGEST_URL",
|
|
config_url or "https://ch1.b.uat.4.cn/knowes/api/knowledge-outer/add",
|
|
)
|
|
|
|
|
|
@router.post("/publish", response_model=KnowledgeIngestPublishResponse, summary="Publish markdown artifact to external knowledge base")
|
|
async def publish_to_knowledge_base(request: KnowledgeIngestPublishRequest) -> KnowledgeIngestPublishResponse:
|
|
title = request.title.strip()
|
|
content = request.content.strip()
|
|
if not title:
|
|
raise HTTPException(status_code=422, detail="标题不能为空")
|
|
if not content:
|
|
raise HTTPException(status_code=422, detail="内容为空，无法入库")
|
|
|
|
try:
|
|
agent_cfg = load_agent_config(request.agent_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
template_name = (agent_cfg.template_name or "").strip() if agent_cfg else ""
|
|
if not template_name:
|
|
raise HTTPException(status_code=400, detail=f"智能体 '{request.agent_id}' 未配置入库模板名（template_name），无法发布")
|
|
|
|
now = datetime.now()
|
|
# nodeName 是入库记录的唯一标识：md 标题 + 时间戳。两次尝试必须共用同一个
|
|
# nodeName，避免第一次实际已入库时第二次产生重复记录。
|
|
timestamp = now.strftime("%Y%m%d%H%M%S")
|
|
node_name = _build_node_name(title, timestamp)
|
|
batch_number = f"md{now.strftime('%Y%m%d')}"
|
|
ingest_url = _resolve_ingest_url()
|
|
|
|
logger.info(
|
|
"[knowledge-ingest] 开始入库: agent=%s template=%s nodeName=%s batchNumber=%s title=%r content_len=%d url=%s",
|
|
request.agent_id,
|
|
template_name,
|
|
node_name,
|
|
batch_number,
|
|
title,
|
|
len(content),
|
|
ingest_url,
|
|
)
|
|
|
|
attempts: list[KnowledgeIngestAttempt] = []
|
|
headers = {"Content-Type": "application/json", "User-Agent": "DeerFlow/1.0"}
|
|
|
|
async with httpx.AsyncClient(timeout=_REQUEST_TIMEOUT_SECONDS) as client:
|
|
for title_field in _TITLE_FIELD_CANDIDATES:
|
|
payload = _build_payload(template_name, batch_number, node_name, title_field, title, content)
|
|
logger.info(
|
|
"[knowledge-ingest] 第 %d 次发送 (标题字段=%s): %s",
|
|
len(attempts) + 1,
|
|
title_field,
|
|
json.dumps({**payload, "list": [{**payload["list"][0], "data": [{"fieldName": d["fieldName"], "value": d["value"][:200]} for d in payload["list"][0]["data"]]}]}, ensure_ascii=False),
|
|
)
|
|
try:
|
|
response = await client.post(ingest_url, json=payload, headers=headers)
|
|
except httpx.HTTPError as exc:
|
|
logger.error("[knowledge-ingest] 第 %d 次发送请求异常 (标题字段=%s): %s", len(attempts) + 1, title_field, exc)
|
|
attempts.append(KnowledgeIngestAttempt(title_field=title_field, error=str(exc)))
|
|
continue
|
|
|
|
text = response.text or ""
|
|
success = _judge_success(response.status_code, text)
|
|
logger.info(
|
|
"[knowledge-ingest] 第 %d 次响应 (标题字段=%s): status=%d success=%s body=%s",
|
|
len(attempts) + 1,
|
|
title_field,
|
|
response.status_code,
|
|
success,
|
|
text[:2000],
|
|
)
|
|
attempts.append(
|
|
KnowledgeIngestAttempt(
|
|
title_field=title_field,
|
|
status_code=response.status_code,
|
|
response_text=text[:2000],
|
|
success=success,
|
|
)
|
|
)
|
|
if success:
|
|
break
|
|
logger.warning("[knowledge-ingest] 第 %d 次入库判定失败 (标题字段=%s)，准备换字段重试", len(attempts), title_field)
|
|
|
|
overall = any(a.success for a in attempts)
|
|
if overall:
|
|
logger.info("[knowledge-ingest] 入库成功: nodeName=%s 使用标题字段=%s", node_name, next(a.title_field for a in attempts if a.success))
|
|
else:
|
|
logger.error("[knowledge-ingest] 入库失败（全部 %d 次尝试均未成功）: nodeName=%s", len(attempts), node_name)
|
|
|
|
return KnowledgeIngestPublishResponse(
|
|
success=overall,
|
|
template_name=template_name,
|
|
node_name=node_name,
|
|
batch_number=batch_number,
|
|
attempts=attempts,
|
|
)
|