Dify 实战：私有化部署与知识库应用

#!/usr/bin/env bash
# =============================================================================
# license-check.sh —— 自建之前先把许可证和活跃度查清楚
#
# 「开源」两个字不是能直接拿来做决策的结论。要回答三个问题：许可证是标准协议
# 还是标准协议 + 附加条款？附加条款限制了什么（多租户转售？前端标识不得移
# 除？）？项目还活着吗？答案会随时间变，所以这里给的是「怎么查」。
#
#   bash license-check.sh langgenius/dify infiniflow/ragflow
#
# 可选 GITHUB_TOKEN 提高速率上限；令牌永远走环境变量，不写进脚本。
# =============================================================================
set -uo pipefail

REPOS=("$@")
[ "${#REPOS[@]}" -eq 0 ] && REPOS=(langgenius/dify infiniflow/ragflow)

AUTH=()
[ -n "${GITHUB_TOKEN:-}" ] && AUTH=(-H "Authorization: Bearer ${GITHUB_TOKEN}")

api() { curl -sS --max-time 20 -H "Accept: application/vnd.github+json" "${AUTH[@]}" "$1"; }

# 没有 jq 时退回 grep/sed，保证在最小化的服务器上也能跑
field() {
  if command -v jq >/dev/null 2>&1; then
    printf '%s' "$1" | jq -r "$2 // \"-\""
  else
    local plain="${2#.}"; plain="${plain%%.*}"
    printf '%s' "$1" | tr ',' '\n' | grep -m1 "\"${plain}\"" \
      | sed 's/.*: *//; s/^"//; s/"$//' || echo "-"
  fi
}

for repo in "${REPOS[@]}"; do
  echo "=============================================================="
  echo "仓库：$repo"
  meta="$(api "https://api.github.com/repos/${repo}")"
  if printf '%s' "$meta" | grep -q '"message": *"Not Found"'; then
    echo "  查不到这个仓库，检查拼写"; continue
  fi

  spdx="$(field "$meta" '.license.spdx_id')"
  echo "  star 数        $(field "$meta" '.stargazers_count')"
  echo "  最后一次提交   $(field "$meta" '.pushed_at')"
  echo "  许可证标识     ${spdx}"
  echo "  许可证名称     $(field "$meta" '.license.name')"

  # 关键一步：标识为 NOASSERTION / Other 意味着这不是标准协议原文，
  # 项目方在标准协议之外加了自己的条款，必须去读 LICENSE 原文。
  case "$spdx" in
    NOASSERTION|Other|-|null)
      echo
      echo "  [注意] 不是标准协议，必读 LICENSE 原文中的附加条款。抓取限制性措辞："
      raw=""
      for branch in main master; do
        raw="$(curl -sS --max-time 20 \
               "https://raw.githubusercontent.com/${repo}/${branch}/LICENSE" || true)"
        [ -n "$raw" ] && break
      done
      if [ -n "$raw" ]; then
        printf '%s\n' "$raw" | grep -n -i -E \
          "multi-tenant|tenant|logo|copyright information|commercial|written permission|may not" \
          | head -n 20 | sed 's/^/           /'
      else
        echo "           抓不到 LICENSE，去仓库页面手动读一遍"
      fi ;;
    *)
      echo
      echo "  标准协议 ${spdx}；仍建议扫一眼 LICENSE 与 NOTICE 是否另有说明。" ;;
  esac
  echo
done

cat <<'TIP'
==============================================================
读数的方法比数本身重要：star 只看得出热度，看不出能不能商用；最后一次
提交超半年，维护成本要按「自己接手」估；标识为 NOASSERTION 等于项目方说
「我改过条款」，必须逐条读。以上均为运行这一刻的快照，要引用就连同查询
时间一起写下来并说明复核办法 —— 过几个月它们一定会变。
TIP

# -*- coding: utf-8 -*-
"""
chunking_preview.py —— 在本地预演分段，看清楚「书被拆成了什么样的卡片」

分段是整条链路里最便宜也最要命的一步：它决定检索能拿回什么，
而检索拿不回来的内容，模型再强也答不出来。
但分段效果在平台上只能一块块翻着看，翻二十块就没耐心了。

纯标准库，不依赖任何服务。给出长度分布、碎块、巨块、被切断的句子 ——
灌库之前就知道参数往哪调。token 按「中文字 1、英文词 1」粗估，够判断量级。

    python chunking_preview.py 劳动合同法.txt --max-tokens 300 --overlap 60
    python chunking_preview.py 劳动合同法.txt --mode hierarchical --dump 3
"""
import argparse
import re
import sys

SENT_END = re.compile(r"(?<=[。！？；!?;])\s*")
CJK = r"[\u3400-\u9fff\u3000-\u303f\uff00-\uffef]"


def est_tokens(text):
    return len(re.findall(CJK, text)) + len(re.sub(CJK, " ", text).split())


def _by_sentence(text, limit):
    """按句子聚成不超过 limit 的块。单段超长时只能这样硬切。"""
    out, cur, tok = [], [], 0
    for s in [x for x in SENT_END.split(text) if x.strip()]:
        n = est_tokens(s)
        if cur and tok + n > limit:
            out.append("".join(cur))
            cur, tok = [], 0
        cur.append(s)
        tok += n
    if cur:
        out.append("".join(cur))
    return out


def split_general(text, sep="\n\n", max_tokens=500, overlap=80):
    """通用分段：按分隔符切开、按上限合并、相邻块留重叠。

    重叠救的是「答案正好横跨两块」：没有重叠时边界上的句子被一刀两断，
    两块各拿半句，谁都匹配不上。
    """
    chunks, buf, tok = [], [], 0
    for piece in [p.strip() for p in text.split(sep) if p.strip()]:
        n = est_tokens(piece)
        if n > max_tokens:                    # 单段就超长，通用分段绕不开的损失
            if buf:
                chunks.append("\n\n".join(buf))
                buf, tok = [], 0
            chunks += _by_sentence(piece, max_tokens)
            continue
        if buf and tok + n > max_tokens:
            chunks.append("\n\n".join(buf))
            buf, tok = [], 0
        buf.append(piece)
        tok += n
    if buf:
        chunks.append("\n\n".join(buf))
    if overlap <= 0 or len(chunks) < 2:
        return chunks
    return [chunks[0]] + [chunks[i - 1][-overlap * 2:] + "\n" + chunks[i]
                          for i in range(1, len(chunks))]


def split_hierarchical(text, sep="\n\n", parent_max=1024, child_max=256):
    """父子分段：父块段落级，父块内部再切子块句子级。返回 [(父块, [子块...])]。

    检索用子块匹配，命中后把它所属的父块整块交给模型 ——
    就是「卡片正面写一句话、背面写着它属于哪一章」的那个背面。
    父块也要有上限，否则一个父块就能把上下文窗口撑爆。
    """
    parents = []
    for para in [p.strip() for p in text.split(sep) if p.strip()]:
        parents += (_by_sentence(para, parent_max)
                    if est_tokens(para) > parent_max else [para])
    return [(p, _by_sentence(p, child_max) or [p]) for p in parents]


def stats(lengths, label):
    if lengths:
        s = sorted(lengths)
        print("  %-6s 数量 %-5d 最短 %-5d 中位 %-5d 最长 %-5d 平均 %.0f"
              % (label, len(s), s[0], s[len(s) // 2], s[-1], sum(s) / len(s)))


def warn_chunks(chunks, limit):
    """三类问题是召回变差的主要来源。"""
    tiny = sum(1 for c in chunks if est_tokens(c) < 30)
    huge = sum(1 for c in chunks if est_tokens(c) > limit * 1.3)
    cut = sum(1 for c in chunks
              if c.strip() and c.strip()[-1] not in "。！？；：」』）】!?;:.)]")
    print("\n  体检：")
    print("    过短块（<30 token）  %d 个 —— 语义不完整，容易被噪声挤掉" % tiny)
    print("    超长块（>上限1.3倍） %d 个 —— 一块混了多个主题，向量被摊平" % huge)
    print("    尾部被切断的块       %d 个 —— 句子断在中间，重叠没设够" % cut)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("path", help="纯文本文件（先把 pdf/docx 转成 txt）")
    ap.add_argument("--mode", choices=["general", "hierarchical"], default="general")
    ap.add_argument("--separator", default="\n\n")
    ap.add_argument("--max-tokens", type=int, default=500)
    ap.add_argument("--overlap", type=int, default=80)
    ap.add_argument("--parent-max", type=int, default=1024)
    ap.add_argument("--child-max", type=int, default=256)
    ap.add_argument("--dump", type=int, default=0)
    args = ap.parse_args()

    text = open(args.path, encoding="utf-8").read()
    print("原文约 %d token，%d 字符\n" % (est_tokens(text), len(text)))

    if args.mode == "general":
        chunks = split_general(text, args.separator, args.max_tokens, args.overlap)
        print("通用分段：上限 %d token，重叠 %d token" % (args.max_tokens, args.overlap))
        stats([est_tokens(c) for c in chunks], "块")
        warn_chunks(chunks, args.max_tokens)
        for i, c in enumerate(chunks[:args.dump], 1):
            print("\n--- 第 %d 块（%d token）---\n%s" % (i, est_tokens(c), c[:400]))
    else:
        pairs = split_hierarchical(text, args.separator, args.parent_max, args.child_max)
        kids = [c for _p, cs in pairs for c in cs]
        print("父子分段：父块上限 %d，子块上限 %d" % (args.parent_max, args.child_max))
        stats([est_tokens(p) for p, _c in pairs], "父块")
        stats([est_tokens(c) for c in kids], "子块")
        print("\n  平均每个父块带 %.1f 个子块" % (len(kids) / max(len(pairs), 1)))
        print("  命中的是子块，送进模型的是父块 —— "
              "子块要短到能精准匹配，父块要长到能把话说完整。")
        warn_chunks(kids, args.child_max)
        for i, (p, cs) in enumerate(pairs[:args.dump], 1):
            print("\n--- 父块 %d（%d token，%d 个子块）---\n%s"
                  % (i, est_tokens(p), len(cs), p[:300]))
    return 0


if __name__ == "__main__":
    sys.exit(main())

# -*- coding: utf-8 -*-
"""
retrieval_eval.py —— 召回质量评测：用一批问题给知识库打分

界面上的召回测试一次只能问一个问题，看到「命中了」很容易自我感觉良好。
但换分段、换 Embedding、开不开 Rerank、TopK 调大调小，到底哪个更好？
必须用同一批问题跑同一套指标，一次只改一个变量。

    hit@k   前 k 条里有没有命中 —— 最该看的，它等于答案质量的上限
    MRR     第一条命中名次的倒数平均 —— 衡量排序，Rerank 的价值在这里
    平均分  只用来给 Score 阈值找起点，不能拿来判断好坏

评测集一行一条 JSON（真实用户问过的最值钱，二三十条就能开始）：
    {"query": "试用期最长能约多久", "doc": "劳动合同法.pdf", "must_contain": "试用期不得超过"}

环境变量：DIFY_BASE_URL / DIFY_DATASET_KEY / DIFY_DATASET_ID

    python retrieval_eval.py evalset.jsonl --top-k 8 --rerank gitee_ai:bge-reranker-v2-m3
    python retrieval_eval.py evalset.jsonl --compare      # 三种检索方式横向对比
"""
import argparse
import json
import os
import sys

import requests

BASE_URL = os.environ.get("DIFY_BASE_URL", "http://127.0.0.1/v1").rstrip("/")
DATASET_KEY = os.environ.get("DIFY_DATASET_KEY")
DATASET_ID = os.environ.get("DIFY_DATASET_ID")

# 界面上分别叫向量检索、全文检索、混合检索
METHODS = ["semantic_search", "full_text_search", "hybrid_search"]


def _headers():
    if not DATASET_KEY or not DATASET_ID:
        raise RuntimeError("缺少 DIFY_DATASET_KEY / DIFY_DATASET_ID")
    return {"Authorization": "Bearer %s" % DATASET_KEY,
            "Content-Type": "application/json"}


def retrieve(query, method="hybrid_search", top_k=4, score=None, rerank=None):
    """调检索接口，返回 [(文档名, 分段文本, 分数)]。就是界面「召回测试」背后那一个。"""
    cfg = {"search_method": method, "reranking_enable": bool(rerank),
           "top_k": top_k, "score_threshold_enabled": score is not None}
    if score is not None:
        cfg["score_threshold"] = score
    if rerank:
        cfg["reranking_model"] = {"reranking_provider_name": rerank[0],
                                  "reranking_model_name": rerank[1]}
    resp = requests.post("%s/datasets/%s/retrieve" % (BASE_URL, DATASET_ID),
                         headers=_headers(), timeout=90,
                         json={"query": query, "retrieval_model": cfg})
    if resp.status_code >= 400:
        raise RuntimeError("HTTP %s: %s" % (resp.status_code, resp.text[:300]))
    return [((s.get("document") or {}).get("name", "?"), s.get("content", ""),
             rec.get("score", 0.0))
            for rec in (resp.json().get("records") or [])
            for s in [rec.get("segment") or {}]]


def judge(case, hits):
    """返回第几条命中，没命中返回 0。

    判定刻意放宽：文档名对上、或关键串出现在分段里都算命中。
    严格的 chunk 级标注要人工投入，起步阶段不值得。
    """
    for rank, (doc, text, _s) in enumerate(hits, 1):
        if case.get("doc") and case["doc"] in doc:
            return rank
        if case.get("must_contain") and case["must_contain"] in text:
            return rank
    return 0


def evaluate(cases, method, top_k, score, rerank, verbose=False):
    n = len(cases)
    hit1 = hit3 = hitk = empty = cnt = 0
    rr_sum = score_sum = 0.0
    for case in cases:
        try:
            hits = retrieve(case["query"], method, top_k, score, rerank)
        except Exception as exc:                          # noqa: BLE001
            print("  [错误] %s -> %s" % (case["query"], exc))
            continue
        empty += not hits
        for _d, _t, sc in hits:
            score_sum += float(sc or 0.0)
            cnt += 1
        rank = judge(case, hits)
        if rank:
            rr_sum += 1.0 / rank
            hitk += 1
            hit3 += rank <= 3
            hit1 += rank == 1
        if verbose:
            print("  %-8s %s" % ("命中@%d" % rank if rank else "未命中", case["query"]))
    return {"样本数": n, "hit@1": hit1 / n if n else 0.0,
            "hit@3": hit3 / n if n else 0.0,
            "hit@%d" % top_k: hitk / n if n else 0.0,
            "MRR": rr_sum / n if n else 0.0,
            "平均相似度": score_sum / cnt if cnt else 0.0, "空召回数": empty}


def load_cases(path):
    cases = []
    for ln, line in enumerate(open(path, encoding="utf-8"), 1):
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            print("  [跳过] 第 %d 行不是合法 JSON" % ln)
            continue
        if "query" in obj:
            cases.append(obj)
    return cases


def show(title, result):
    print("\n== %s ==" % title)
    for k, v in result.items():
        print("  %-12s %s" % (k, "%.3f" % v if isinstance(v, float) else v))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("evalset")
    ap.add_argument("--method", choices=METHODS, default="hybrid_search")
    ap.add_argument("--top-k", type=int, default=4)
    ap.add_argument("--score", type=float, default=None, help="Score 阈值；不传表示不启用")
    ap.add_argument("--rerank", default="off", help="off 或 '供应商:模型名'")
    ap.add_argument("--compare", action="store_true")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args()

    rerank = None
    if args.rerank != "off":
        if ":" not in args.rerank:
            print("--rerank 格式是 供应商:模型名")
            return 2
        rerank = tuple(args.rerank.split(":", 1))

    cases = load_cases(args.evalset)
    if not cases:
        print("评测集是空的，先攒 20 条真实问题再来")
        return 1
    print("评测集 %d 条，TopK=%d，Rerank=%s，Score 阈值=%s"
          % (len(cases), args.top_k, args.rerank,
             args.score if args.score is not None else "未启用"))

    if args.compare:
        # 横向对比时唯一变量只能是检索方式，其余参数必须一致，否则结论不成立
        for m in METHODS:
            show(m, evaluate(cases, m, args.top_k, args.score, rerank, args.verbose))
        print("\n看 hit@k 选检索方式，看 MRR 判断要不要上 Rerank。")
        print("差距小于两三个百分点就别换 —— 样本这么小，那是噪声不是改进。")
    else:
        show(args.method, evaluate(cases, args.method, args.top_k,
                                   args.score, rerank, args.verbose))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# =============================================================================
# compose-up.sh —— 用 Docker Compose 把服务拉起来，并自证「它真的起来了」
#
# 两条纪律写进脚本，而不是写在文档里靠人记：
#   1. .env 只在不存在时生成，绝不覆盖已有配置（二次部署最容易把口令盖回默认值）
#   2. 端口通过环境变量覆盖，不改 compose 文件（同机部署第二套时不用回头改）
#
#   DIFY_DOCKER_DIR=/opt/dify/docker DIFY_WEB_PORT=8080 bash compose-up.sh
# =============================================================================
set -euo pipefail

DIFY_DOCKER_DIR="${DIFY_DOCKER_DIR:-$(pwd)}"
DIFY_WEB_PORT="${DIFY_WEB_PORT:-80}"
WAIT_SECONDS="${WAIT_SECONDS:-180}"

log() { printf '[%s] %s\n' "$(date '+%H:%M:%S')" "$*"; }
die() { printf '[FATAL] %s\n' "$*" >&2; exit 1; }

# --- 0. 前置条件。注意是 `docker compose`（v2 插件），不是老的 docker-compose ---
command -v docker >/dev/null 2>&1 || die "没有 docker"
docker compose version >/dev/null 2>&1 || die "没有 compose v2 插件"
docker info >/dev/null 2>&1 || die "docker daemon 没跑起来"

cd "$DIFY_DOCKER_DIR" || die "目录不存在：$DIFY_DOCKER_DIR"
[ -f docker-compose.yaml ] || [ -f docker-compose.yml ] \
  || die "这里没有 compose 文件；要进的是仓库里的 docker/ 子目录，不是仓库根目录"

# --- 1. .env：只补不盖 ---
if [ -f .env ]; then
  log ".env 已存在，保持原样"
else
  [ -f .env.example ] || die "既没有 .env 也没有 .env.example"
  cp .env.example .env
  log "已从 .env.example 生成 .env —— 首次部署请先改 SECRET_KEY 与数据库口令"
fi

if [ "$DIFY_WEB_PORT" != "80" ]; then
  export EXPOSE_NGINX_PORT="$DIFY_WEB_PORT"
  log "Web 端口覆盖为 $DIFY_WEB_PORT（环境变量方式，不改 compose 文件）"
fi

# --- 2. 起服务。不加 --force-recreate，避免无谓重建已经健康的容器 ---
log "开始拉取镜像并启动（首次会比较慢，镜像体积以 GB 计）"
docker compose up -d

# --- 3. 验活：轮询到真能响应为止 ---
# 「容器 Up」不等于「服务可用」：迁移数据库、初始化向量库都要时间。
url="http://127.0.0.1:${DIFY_WEB_PORT}/install"
log "等待服务就绪：$url"
deadline=$(( $(date +%s) + WAIT_SECONDS ))
code=000
while [ "$(date +%s)" -lt "$deadline" ]; do
  code="$(curl -s -o /dev/null -w '%{http_code}' --max-time 5 "$url" || echo 000)"
  case "$code" in 200|302|307) log "服务已就绪，HTTP $code"; break ;; esac
  sleep 5
done

echo; log "容器状态："; docker compose ps

case "$code" in
  200|302|307) ;;
  *)
    cat <<'TIP'

[WARN] 超时仍未拿到正常响应。按这个顺序查，不要瞎重启：
  1) docker compose ps                    有没有容器 Restarting / Exited
  2) docker compose logs --tail=120 api    后端报错基本都在这里
  3) docker compose logs --tail=80  db     改过口令但没清卷，会一直认证失败
  4) ss -ltnp | grep ':80 '                 宿主机端口被别的服务占了
TIP
    exit 1 ;;
esac

cat <<TIP

[OK] 打开 http://127.0.0.1:${DIFY_WEB_PORT}/install 完成首次管理员注册。
日常命令：ps 查状态；logs -f api 跟踪日志；stop / down 保留数据卷；
  docker compose down -v       连数据卷一起删 —— 知识库与向量数据一并消失，慎用
TIP

#!/usr/bin/env bash
# =============================================================================
# backup-volumes.sh —— 私有化部署的命门：把数据卷备份出来
#
# 容器可以随便删了重建，但这几类数据删掉就没了：
#   关系数据库（应用配置、工作流版本、会话）、向量数据（重灌要重花 Embedding 算力）、
#   对象存储（上传的原始文档）。`docker compose down -v` 一条命令就能全部抹掉。
#
# 做法：不直接 tar 宿主机目录（named volume 不在 compose 目录下），
# 而是起一个临时容器把卷挂进去打包 —— 与卷的存储位置无关，两种挂载都适用。
#
#   bash backup-volumes.sh
#   BACKUP_DIR=/data/backup VOLUME_FILTER=dify bash backup-volumes.sh
# =============================================================================
set -euo pipefail

BACKUP_DIR="${BACKUP_DIR:-$(pwd)/dify-backup}"
VOLUME_FILTER="${VOLUME_FILTER:-dify}"
OUT="${BACKUP_DIR}/$(date '+%Y%m%d-%H%M%S')"

command -v docker >/dev/null 2>&1 || { echo "[FATAL] 没有 docker" >&2; exit 1; }
mkdir -p "$OUT"

mapfile -t VOLUMES < <(docker volume ls --format '{{.Name}}' | grep -i "$VOLUME_FILTER" || true)
if [ "${#VOLUMES[@]}" -eq 0 ]; then
  echo "[FATAL] 没匹配到数据卷（过滤词：$VOLUME_FILTER），先看 docker volume ls" >&2
  exit 1
fi

echo "将备份 ${#VOLUMES[@]} 个数据卷到 $OUT"
printf '  - %s\n' "${VOLUMES[@]}"

# 备份期间建议停服，保证落盘一致。不停机就至少对数据库单独做一次逻辑导出。
if [ "${STOP_FIRST:-1}" = "1" ]; then
  echo "先停服务以保证一致（STOP_FIRST=0 可跳过，但备份可能不一致）"
  docker compose stop || true
fi

for v in "${VOLUMES[@]}"; do
  echo "打包 $v ..."
  docker run --rm -v "${v}:/src:ro" -v "${OUT}:/dst" alpine:3 \
    tar czf "/dst/${v}.tar.gz" -C /src .
  echo "  -> ${v}.tar.gz  $(du -h "${OUT}/${v}.tar.gz" | cut -f1)"
done

[ "${STOP_FIRST:-1}" = "1" ] && { echo "恢复服务"; docker compose start || true; }

# 备份完立刻验，否则等到要恢复时才发现是个空壳就晚了。
echo
echo "== 校验归档完整性 =="
fail=0
for f in "$OUT"/*.tar.gz; do
  if tar tzf "$f" >/dev/null 2>&1; then
    printf '  [OK]   %-44s %s 个条目\n' "$(basename "$f")" "$(tar tzf "$f" | wc -l)"
  else
    printf '  [BAD]  %s 无法读取\n' "$(basename "$f")"; fail=1
  fi
done
( cd "$OUT" && sha256sum ./*.tar.gz > SHA256SUMS )
echo "  校验和已写入 $OUT/SHA256SUMS"

cat <<TIP

恢复办法（目标机器上，卷为空时）：
  docker volume create <卷名>
  docker run --rm -v <卷名>:/dst -v ${OUT}:/src alpine:3 \\
    tar xzf /src/<卷名>.tar.gz -C /dst

两地各存一份：服务器本地一份，再拉回另一台机器一份。
只存在同一台机器上的备份，在这台机器出事时等于没有。
TIP
exit "$fail"

# -*- coding: utf-8 -*-
"""
kb_bulk_upload.py —— 批量把一个目录里的文档灌进知识库

界面上传一次只能选有限几个文件，几百份规章制度靠手点不现实。
更重要的是：走脚本才能保证「同一批文档用同一套分段参数」。
手点最容易出现的事故是前五十个用父子分段、后五十个忘了改，
召回效果差一大截却查不出原因。

环境变量：DIFY_BASE_URL / DIFY_DATASET_KEY（知识库密钥，不是应用密钥）/ DIFY_DATASET_ID

    python kb_bulk_upload.py ./docs --mode hierarchical
    python kb_bulk_upload.py ./docs --dry-run

这是写操作，先在空知识库上试，别直接对生产库跑。
"""
import argparse
import json
import os
import sys
import time

import requests

BASE_URL = os.environ.get("DIFY_BASE_URL", "http://127.0.0.1/v1").rstrip("/")
DATASET_KEY = os.environ.get("DIFY_DATASET_KEY")
DATASET_ID = os.environ.get("DIFY_DATASET_ID")
SUPPORTED = (".txt", ".md", ".mdx", ".pdf", ".html", ".htm",
             ".docx", ".csv", ".xlsx", ".xls")

_PRE = [{"id": "remove_extra_spaces", "enabled": True},
        {"id": "remove_urls_emails", "enabled": False}]

# 两种分段规则。差别只在 parent_mode 与有没有 subchunk_segmentation，
# 但对召回效果的影响，比换一个更贵的模型大得多。
RULES = {
    # 通用分段：一刀切成等长块。separator 优先按空行切，尽量不切断自然段；
    # chunk_overlap 约为上限的 16%，落在 10~25% 的经验区间内。
    "general": {"indexing_technique": "high_quality", "process_rule": {
        "mode": "custom", "rules": {"pre_processing_rules": _PRE, "segmentation": {
            "separator": "\n\n", "max_tokens": 500, "chunk_overlap": 80}}}},
    # 父子分段：父块按段落（full_doc 表示整篇当父块），子块切到句子级
    "hierarchical": {"indexing_technique": "high_quality",
                     "doc_form": "hierarchical_model", "process_rule": {
        "mode": "hierarchical", "rules": {
            "pre_processing_rules": _PRE, "parent_mode": "paragraph",
            "segmentation": {"separator": "\n\n", "max_tokens": 1024},
            "subchunk_segmentation": {"separator": "\n", "max_tokens": 256}}}},
}


def _headers():
    if not DATASET_KEY or not DATASET_ID:
        raise RuntimeError("缺少 DIFY_DATASET_KEY / DIFY_DATASET_ID")
    return {"Authorization": "Bearer %s" % DATASET_KEY}


def collect(folder):
    ok, skipped = [], []
    for root, _dirs, files in os.walk(folder):
        for name in sorted(files):
            path = os.path.join(root, name)
            if os.path.splitext(name)[1].lower() not in SUPPORTED:
                skipped.append((path, "格式不支持"))
            elif os.path.getsize(path) == 0:
                skipped.append((path, "空文件"))
            else:
                ok.append(path)
    return ok, skipped


def upload_one(path, rule):
    """multipart 上传：文件走 files，分段规则走名为 data 的表单字段（JSON 字符串）。

    把规则塞进 json= 是最常见的写法错误：接口收下了文件却用默认分段规则，且不报错。
    """
    with open(path, "rb") as fh:
        resp = requests.post(
            "%s/datasets/%s/document/create-by-file" % (BASE_URL, DATASET_ID),
            headers=_headers(), timeout=300,
            files={"file": (os.path.basename(path), fh)},
            data={"data": json.dumps(rule, ensure_ascii=False)})
    if resp.status_code >= 400:
        raise RuntimeError("HTTP %s: %s" % (resp.status_code, resp.text[:300]))
    body = resp.json()
    return (body.get("document") or {}).get("id"), body.get("batch")


def wait_indexing(batch, poll=5, limit=600):
    """等这一批完成索引。

    「上传成功」不等于「可以检索」：文本要切块，再逐块过 Embedding。
    灌完立刻做召回测试只会得到空结果，然后误以为是分段参数配错了。
    """
    url = "%s/datasets/%s/documents/%s/indexing-status" % (BASE_URL, DATASET_ID, batch)
    waited = 0
    while waited < limit:
        resp = requests.get(url, headers=_headers(), timeout=60)
        if resp.status_code >= 400:
            return "查询失败 HTTP %s" % resp.status_code
        items = resp.json().get("data") or []
        done = sum(1 for i in items if i.get("indexing_status") == "completed")
        bad = [i for i in items if i.get("indexing_status") == "error"]
        if bad:
            return "有 %d 个文档索引失败：%s" % (len(bad), bad[0].get("error") or "原因未返回")
        if items and done == len(items):
            return "全部完成（%d 个）" % done
        print("    索引中 %d/%d ..." % (done, len(items)))
        time.sleep(poll)
        waited += poll
    return "等待超时，去知识库页面看具体进度"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("folder")
    ap.add_argument("--mode", choices=list(RULES), default="hierarchical")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--sleep", type=float, default=1.0, help="每个文件间隔秒数")
    args = ap.parse_args()

    files, skipped = collect(args.folder)
    print("待上传 %d 个，跳过 %d 个，分段方式：%s" % (len(files), len(skipped), args.mode))
    for path, why in skipped:
        print("  [跳过] %s （%s）" % (path, why))
    if args.dry_run:
        for p in files:
            print("  [待传] %s" % p)
        return 0
    if not files:
        return 1

    rule, last_batch, failed = RULES[args.mode], None, []
    for i, path in enumerate(files, 1):
        print("[%d/%d] %s" % (i, len(files), os.path.basename(path)))
        try:
            doc_id, batch = upload_one(path, rule)
            last_batch = batch or last_batch
            print("    已提交，文档 id=%s" % doc_id)
        except Exception as exc:                          # noqa: BLE001
            print("    失败：%s" % exc)
            failed.append((path, str(exc)))
        time.sleep(args.sleep)

    if last_batch:
        print("\n等待索引完成 ...\n  %s" % wait_indexing(last_batch))
    print("\n完成：成功 %d，失败 %d" % (len(files) - len(failed), len(failed)))
    for path, why in failed:
        print("  [失败] %s -> %s" % (path, why))
    print("\n下一步：跑一次召回评测，别凭感觉认为灌好了。")
    return 0 if not failed else 1


if __name__ == "__main__":
    sys.exit(main())

# =============================================================================
# wf_kb_minimal.yml —— 最小的「知识库问答」对话流（chatflow）
#
#     开始  →  知识检索  →  LLM  →  直接回复
#
# 可以直接用「导入 DSL 文件」创建应用。导入前改两处（已用 TODO 标出）：
#   TODO-1  dataset_ids：换成自己知识库的 ID（知识库详情页地址栏里那串 UUID）
#   TODO-2  模型：换成自己已经接好的推理模型与 Rerank 模型
#           没接 Rerank 就把 reranking_enable 改成 false
#
# 读 DSL 的价值：界面上的每个开关，在这里都是一个可 diff、可进版本库、
# 可在多套环境之间搬运的字段。界面会改版，字段名相对稳定。
# =============================================================================
app:
  description: 最小知识库问答：检索命中的分段作为上下文交给模型作答
  mode: advanced-chat        # 对话流（带会话上下文）；workflow 是普通工作流
  name: 知识库问答（最小版）
kind: app
version: 0.1.5

workflow:
  features:
    file_upload: {enabled: false}
    opening_statement: 你好，这里是私有知识库问答。问题越具体，检索越准。
    # 开启后回答下方会列出「引用了哪些分段」，排查召回问题全靠它
    retriever_resource: {enabled: true}

  graph:
    # 连线用节点 id，不是标题。改节点时连线要一起改，
    # 否则画布上会出现孤儿节点，而流程静默跑不通。
    edges:
      - {id: e1, source: 'node_start',     sourceHandle: source, target: 'node_retrieval', type: custom}
      - {id: e2, source: 'node_retrieval', sourceHandle: source, target: 'node_llm',       type: custom}
      - {id: e3, source: 'node_llm',       sourceHandle: source, target: 'node_answer',    type: custom}

    nodes:
      # 对话流入口。用户这一轮说的话由系统变量 sys.query 承载，不用自己再声明；
      # variables 里加的是额外表单项（比如让用户选部门、选文档范围）。
      - id: 'node_start'
        type: custom
        position: {x: 30, y: 250}
        data: {title: 开始, type: start, variables: []}

      # 这一页的主角。
      #   query_variable_selector  用哪个变量去检索
      #   retrieval_mode: multiple 多路召回
      #   top_k / reranking_enable 召回条数与是否重排
      # 注意：TopK 与 Score 阈值只在 Rerank 这一步生效。不开 Rerank 就调这两个值，
      #       调了也不会按预期起作用 —— 这是最容易浪费半小时的地方。
      - id: 'node_retrieval'
        type: custom
        position: {x: 334, y: 250}
        data:
          title: 知识检索
          type: knowledge-retrieval
          dataset_ids: ['TODO-1-换成你自己的知识库-ID']
          query_variable_selector: ['node_start', 'sys.query']
          retrieval_mode: multiple
          multiple_retrieval_config:
            top_k: 4
            reranking_enable: true
            reranking_mode: reranking_model
            reranking_model: {provider: 'TODO-2-Rerank-供应商', model: 'TODO-2-Rerank-模型'}

      # 把检索结果当上下文，让模型写答案。
      # context.variable_selector 指向检索节点的 result，提示词里的 {{#context#}}
      # 才会被替换成召回的分段。只连线、不在 context 里挂上 result，占位符就是空的 ——
      # 现象是「模型一本正经地瞎答」，而画布上看起来一切正常。
      - id: 'node_llm'
        type: custom
        position: {x: 638, y: 250}
        data:
          title: LLM
          type: llm
          model:
            provider: 'TODO-2-推理模型供应商'
            name: 'TODO-2-推理模型名'
            mode: chat
            completion_params: {temperature: 0.3}   # 知识库问答要忠于原文，温度调低
          context: {enabled: true, variable_selector: ['node_retrieval', 'result']}
          prompt_template:
            - role: system
              text: |
                你是一个严谨的知识库问答助手。
                只依据下面提供的资料回答问题，先给结论，再给出资料原文作为依据。
                资料中没有提到的内容，直接回答「资料中没有相关内容」，不要自行推测。

                资料：
                {{#context#}}
            - role: user
              text: '{{#sys.query#}}'
          vision: {enabled: false}

      # 把 LLM 的输出吐给用户。引用的是「节点 id + 变量名」，改了 id 这里要同步改。
      - id: 'node_answer'
        type: custom
        position: {x: 942, y: 250}
        data: {title: 直接回复, type: answer, answer: '{{#node_llm.text#}}'}

    viewport: {x: 0, y: 0, zoom: 0.8}

# -*- coding: utf-8 -*-
"""
dify_chat_api.py —— 用 HTTP API 调用已经发布的应用

界面上搭好的对话流，本质上就是一个 HTTP 接口。接进自己的业务系统只需要三样：
服务地址（私有化部署就是自己那台机器）、应用密钥、用户标识（用来隔离会话）。

两种应用类型对应两组接口，路径和参数不同，别写混：
    对话型（chatflow / agent / 聊天助手）   POST /v1/chat-messages
    工作流型（workflow）                    POST /v1/workflows/run

环境变量：DIFY_BASE_URL（要带 /v1）、DIFY_API_KEY

    pip install requests
    python dify_chat_api.py "试用期最长能约多久"
"""
import json
import os
import sys
import uuid

import requests

BASE_URL = os.environ.get("DIFY_BASE_URL", "http://127.0.0.1/v1").rstrip("/")
API_KEY = os.environ.get("DIFY_API_KEY")
TIMEOUT = int(os.environ.get("DIFY_TIMEOUT", "120"))


def _headers():
    if not API_KEY:
        raise RuntimeError("缺少 DIFY_API_KEY，先在应用的访问 API 处生成密钥")
    return {"Authorization": "Bearer %s" % API_KEY, "Content-Type": "application/json"}


def chat_blocking(query, user, conversation_id="", inputs=None):
    """阻塞模式：一次拿到完整答案。写脚本、跑批、做评测时最省事，前端直接用会显得卡。

    conversation_id 传空表示新会话；把上一轮返回的值原样传回来就能接着聊 ——
    记忆靠它，不要自己拼历史。
    """
    resp = requests.post("%s/chat-messages" % BASE_URL, headers=_headers(),
                         timeout=TIMEOUT,
                         json={"inputs": inputs or {},   # 开始节点的自定义表单变量
                               "query": query,          # 对应 sys.query
                               "response_mode": "blocking",
                               "conversation_id": conversation_id,
                               "user": user})           # 必填，同一 user 的会话才互相可见
    # 出错时先把响应体打出来 —— 只看状态码排查不了「知识库没挂上」这类问题
    if resp.status_code >= 400:
        raise RuntimeError("HTTP %s: %s" % (resp.status_code, resp.text[:500]))
    data = resp.json()
    return data.get("answer", ""), data.get("conversation_id", ""), data


def chat_streaming(query, user, conversation_id="", inputs=None, on_chunk=None):
    """流式模式：按 SSE 逐块推送，前端体验好。事件类型不止一种，按需扩展。"""
    parts, conv_id, resources = [], conversation_id, []
    with requests.post("%s/chat-messages" % BASE_URL, headers=_headers(),
                       stream=True, timeout=TIMEOUT,
                       json={"inputs": inputs or {}, "query": query,
                             "response_mode": "streaming",
                             "conversation_id": conversation_id,
                             "user": user}) as resp:
        if resp.status_code >= 400:
            raise RuntimeError("HTTP %s: %s" % (resp.status_code, resp.text[:500]))
        for raw in resp.iter_lines(decode_unicode=True):
            if not raw or not raw.startswith("data:"):
                continue
            chunk = raw[5:].strip()
            if not chunk or chunk == "[DONE]":
                continue
            try:
                evt = json.loads(chunk)
            except json.JSONDecodeError:
                continue                      # 半截的包跳过，下一轮会补齐

            etype = evt.get("event")
            if etype in ("message", "agent_message"):
                parts.append(evt.get("answer", ""))
                if on_chunk:
                    on_chunk(evt.get("answer", ""))
            elif etype == "message_end":
                conv_id = evt.get("conversation_id", conv_id)
                # 引用了哪些分段藏在这里：排查召回问题第一个要看的字段
                resources = (evt.get("metadata") or {}).get("retriever_resources") or []
            elif etype == "error":
                raise RuntimeError("流式返回错误：%s" % evt.get("message"))
    return "".join(parts), conv_id, resources


def run_workflow(inputs, user, response_mode="blocking"):
    """工作流型：没有 query，只有 inputs；入参名就是开始节点里定义的变量名。

    返回结构也不同：结果在 data.outputs 里。
    """
    resp = requests.post("%s/workflows/run" % BASE_URL, headers=_headers(),
                         timeout=TIMEOUT,
                         json={"inputs": inputs, "response_mode": response_mode,
                               "user": user})
    if resp.status_code >= 400:
        raise RuntimeError("HTTP %s: %s" % (resp.status_code, resp.text[:500]))
    data = resp.json()
    return (data.get("data") or {}).get("outputs", {}), data


def print_sources(resources):
    """私有知识库问答的可信度，全靠「答案能溯源到哪一段原文」撑着。"""
    if not resources:
        print("\n[没有引用任何分段] —— 要么没命中，要么根本没挂知识库")
        return
    print("\n== 引用的分段（%d 条）==" % len(resources))
    for i, r in enumerate(resources, 1):
        print("  %d. %s   相似度：%s" % (i, r.get("document_name", "?"), r.get("score", "?")))
        text = (r.get("content") or "").replace("\n", " ")
        print("     %s%s" % (text[:110], "…" if len(text) > 110 else ""))


def main():
    question = sys.argv[1] if len(sys.argv) > 1 else "介绍一下你能做什么"
    user = os.environ.get("DIFY_USER") or "debug-%s" % uuid.uuid4().hex[:8]
    print("提问：%s\n\n回答：" % question, end="", flush=True)
    _answer, conv_id, resources = chat_streaming(
        question, user=user, on_chunk=lambda s: print(s, end="", flush=True))
    print()
    print_sources(resources)
    print("\n会话 id：%s（下一轮传回去就能接上上下文）" % conv_id)


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""
law_prompts.py —— 法律助手四个模块的提示词，集中放在一个文件里

写在画布节点里的提示词，改一个字要开浏览器、找节点、点开、改、发布：
没有 diff、没有版本、没法回滚，出了问题说不清是哪一版改坏的。
收进一个文件后，改动能进版本库、能评审、能回滚，再同步回节点。

它同时是一份「提示词怎么写」的样本：角色 / 任务 / 输出格式 / 约束条件
四段式；必写「资料里没有就说没有」（抑制编造最有效的一句）；检索结果用
{{#context#}} 占位，让提示词与检索解耦。

    python law_prompts.py                 # 列出所有模块
    python law_prompts.py contract_labor  # 打印某个模块的完整提示词
"""
import sys

CONTEXT = "{{#context#}}"          # 检索节点的输出会替换到这里
USER_QUERY = "{{#sys.query#}}"     # 用户这一轮说的话
DOC_TEXT = "{{#文档提取器.text#}}"  # 上传文件解析出的文本


# --- 模块一：法律咨询 -------------------------------------------------------
# 先改写再检索：用户原话里的情绪和铺垫会在向量空间里把方向带偏。
QUERY_REWRITE = {
    "title": "法律咨询 · 查询改写",
    "system": "你是一个法律条文查询助手，能够根据用户的提问进行优化，"
              "转化成更适合查询知识库的问题。只输出改写后的问题本身，不要解释。",
    "user": "用户输入：\n%s" % USER_QUERY,
    "note": "必须开记忆：用户说「那第二种情况呢」时，没有上文就改写不出可检索的问题。",
}

LEGAL_QA = {
    "title": "法律咨询 · 条文作答",
    "system": "你是一个法律咨询助手，能够根据用户的提问和从知识库里查询出来的结果回答用户。"
              "回答时先直接给出结论，再给出对应的法律原文依据。"
              "知识库中没有对应条文时，明确告知未检索到，不要凭印象作答。",
    "user": "用户的问题：\n%s\n知识库：\n%s" % (USER_QUERY, CONTEXT),
    "note": "检索要开 Rerank 并把召回条数压小 —— 用户要的是准确的那一两条原文。",
}


# --- 模块二：案件分析 -------------------------------------------------------
CASE_QUERY_EXTRACT = {
    "title": "案件分析 · 检索条件提取",
    "system": "你是一个法律专家，能够根据用户上传的案例内容和用户的输入拆分出"
              "对应的法律条文查询条件，接下来我将进行知识库的查询。",
    "user": "用户输入：\n%s\n用户上传的案例：\n%s" % (USER_QUERY, DOC_TEXT),
    "note": "案情往往几千字，整段拿去检索等于把意思平均成一个点；先提炼再检索。",
}

CASE_ANALYSIS = {
    "title": "案件分析 · 出具分析报告",
    "system": """你是一名专业的案情分析专家，名为「案理」。

一、核心定位
· 角色：专业、中立、严谨的法律分析助手，梳理案件事实、识别争议焦点、结合条文分析。
· 底线：所有分析严格基于用户提供的材料和法律条文，不臆断或虚构；
        不替代律师提供正式法律意见，不预测判决结果；
        信息不足或存在矛盾时，明确说明局限性。

二、输出结构（按此顺序）
1. 案件基本信息：当事人、案由、时间、核心事实
2. 争议焦点：归纳核心争议点
3. 事实与证据分析：梳理法律事实，评估证据链完整性
4. 法律适用分析：结合条文分析行为性质、责任及后果
5. 综合结论与建议：分析意见与后续行动方向

三、表达要求
· 专业清晰的法律语言，避免「我认为」等主观表述，关键结论加粗
· 引用条文须注明名称与条款序号；发现信息矛盾或依据不足时单独说明""",
    "user": "用户输入：\n%s\n用户上传的案例：\n%s\n\n参考法律条文：\n%s"
            % (USER_QUERY, DOC_TEXT, CONTEXT),
    "note": "「不臆断或虚构」「信息不足需说明局限性」是整份提示词里最值钱的两句。",
}


# --- 模块三：合同分析 -------------------------------------------------------
# 三支共用同一套用户提示词，只有 system 不同 —— 分流分的就是这一段。
_CONTRACT_USER = "用户的案情：\n%s\n用户提供的输入：\n%s\n法律条文：\n%s" % (
    DOC_TEXT, USER_QUERY, CONTEXT)

CONTRACT_LABOR = {
    "title": "合同分析 · 劳动合同",
    "system": """# 角色
你是一名资深劳动法律师，帮助企业审查劳动合同的合规性，规避用工法律风险。

# 工作流
1. 核对合同是否包含法定必备条款，明确指出缺失项及其风险。
2. 重点审查：试用期期限与工资标准是否合法；工作地点与岗位调整条款是否过于宽泛；
   劳动报酬是否明确工资构成、支付时间、加班费基数；解除条件是否合法、补偿与
   赔偿计算是否清晰；竞业限制补偿金是否有明确约定。

# 输出格式
一、总体评价　二、分项分析与修改建议（表格：问题条款 / 风险等级 / 法律依据 /
风险分析 / 修改建议文本）　三、核心风险摘要（3-5 个）　四、谈判策略建议

# 约束条件
- 严禁编造不存在的法条，所有判断必须注明依据出处。
- 合法性存在普遍争议的条款应予提示，并建议咨询执业律师。
- 以下文提供的法律条文为主。""",
    "user": _CONTRACT_USER,
}

CONTRACT_TECH = {
    "title": "合同分析 · 技术服务合同",
    "system": """# 角色
你是科技公司的法务顾问，审查技术服务协议，控制风险、保障项目交付。

# 工作流
1. 工作范围与交付物是否清晰、具体、可量化、可验证。
2. 知识产权：背景知识产权与项目中产生的知识产权归属、许可范围、期限与地域。
3. 付款条件：付款节点是否与关键里程碑强关联。
4. 保密与合规：保密信息定义与期限，数据安全与个人信息保护要求。
5. 违约责任是否对等，责任上限是否合理。

# 输出格式
一、总体风险评估　二、关键条款深度剖析　三、谈判优先级清单
（必须修改 / 建议修改 / 可接受）

# 约束条件
- 基于合同法相关规定，同时侧重商业实践的合理性与风险控制。
- 修改建议应具可操作性，并给出修改后的范例文本。
- 以下文提供的法律条文为主。""",
    "user": _CONTRACT_USER,
}

# 分类判不出类型时走这一支。任何分流都必须有兜底分支，
# 否则遇到没预料到的输入，流程会直接断在这里。
CONTRACT_GENERAL = {
    "title": "合同分析 · 通用兜底",
    "system": """# 角色
你是一名擅长设计风险隔离方案的法律风险控制专家。

# 工作流
1. 兜底事项描述是否清晰无歧义。　2. 责任触发条件是否客观可验证。
3. 责任性质与金额上限是否明确。　4. 是否赋予核查权与追偿权。

# 输出格式
一、协议效力评估　二、核心条款风险提示　三、修订建议与文本

# 约束条件
- 重点提示无限责任风险，强调约定不明可能导致条款无效。
- 以下文提供的法律条文为主。""",
    "user": _CONTRACT_USER,
}


# --- 模块四：资费查询 -------------------------------------------------------
# 上下文来自数据库而不是知识库：结构与前三路一模一样，只是灌进去的料换了来源。
PRICING = {
    "title": "资费查询 · 报价区间",
    "system": "你是一名专业的法律助手，负责律所的价格咨询业务，"
              "能够根据用户输入的问题，结合律所定价细则，给出一个报价区间。"
              "定价表里没有的服务，如实说明需要人工确认，不要自行估算。",
    "user": "用户问题：\n%s\n\n定价：\n%s" % (USER_QUERY, CONTEXT),
    "note": "报价是会产生纠纷的输出，末尾必须挂人工确认的出口。",
}

# 每个输出节点末尾都要挂的边界说明。对外产品里这不是客套话，是必需品。
DISCLAIMER = {
    "legal_qa": "以上内容来自律所智能大脑，建议联系律所获取专业的咨询建议。",
    "case": "以上分析结果仅供参考，建议联系律所获取专业的咨询建议。",
    "contract": "以上分析内容来自律所智能大脑，建议联系律所获取专业的咨询建议。",
    "pricing": "具体报价请联系人工客服确认。",
}

ALL_MODULES = {
    "query_rewrite": QUERY_REWRITE, "legal_qa": LEGAL_QA,
    "case_query": CASE_QUERY_EXTRACT, "case_analysis": CASE_ANALYSIS,
    "contract_labor": CONTRACT_LABOR, "contract_tech": CONTRACT_TECH,
    "contract_general": CONTRACT_GENERAL, "pricing": PRICING,
}


def main():
    if len(sys.argv) > 1:
        mod = ALL_MODULES.get(sys.argv[1])
        if not mod:
            print("没有这个模块。可选：%s" % ", ".join(ALL_MODULES))
            return 1
        print("== %s ==\n\n[system]\n%s\n\n[user]\n%s"
              % (mod["title"], mod["system"], mod["user"]))
        if mod.get("note"):
            print("\n[要点]\n%s" % mod["note"])
        return 0

    print("共 %d 个提示词模块：\n" % len(ALL_MODULES))
    for key, mod in ALL_MODULES.items():
        print("  %-18s %s" % (key, mod["title"]))
    print("\n查看某一个：python law_prompts.py contract_labor")
    return 0


if __name__ == "__main__":
    sys.exit(main())

-- =============================================================================
-- law_assistant_schema.sql —— 律所资费查询模块的业务库
--
-- 这张表是「资费查询」模块唯一的事实来源，名目与数字均为演示用示例。它刻意不放进知识库：
-- 价格是结构化数据，天天在变，灌进向量库就得每次改价都重新分段、重新
-- Embedding，还答不准。结构化的事实查数据库，非结构化的条文查知识库。
--
-- 执行：mysql -h <host> -u <user> -p < law_assistant_schema.sql
-- 口令不写进这个文件，也不写进任何提交到仓库的文件。
-- =============================================================================
CREATE DATABASE IF NOT EXISTS law_assistant
  DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_0900_ai_ci;
USE law_assistant;

-- billing_unit 用枚举而不是自由文本：上下文越规整，模型算出来的报价越稳。
-- is_active 保留下架服务的记录，不做物理删除 —— 报价争议时要能追溯。
CREATE TABLE IF NOT EXISTS legal_service_pricing (
    service_id          INT AUTO_INCREMENT PRIMARY KEY,
    service_name        VARCHAR(255)   NOT NULL COMMENT '服务名称',
    service_description TEXT                    COMMENT '服务说明，模型据此判断用户问的是哪一项',
    price               DECIMAL(10, 2) NOT NULL COMMENT '单价',
    billing_unit        ENUM('按小时', '按次', '按项目') NOT NULL COMMENT '计费单位',
    lawyer_id           INT            NOT NULL COMMENT '负责律师',
    is_active           BOOLEAN        DEFAULT TRUE COMMENT '是否在售',
    created_at          TIMESTAMP      DEFAULT CURRENT_TIMESTAMP,
    updated_at          TIMESTAMP      DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    KEY idx_active (is_active),
    KEY idx_name (service_name)
) ENGINE=InnoDB COMMENT='律所服务定价';

-- 字段注释与中文服务名一起构成了模型的「阅读材料」。
-- 服务名如果写成 SVC_001 这种代号，模型就没法把用户口语里的「打官司」
-- 对应到「民事诉讼一审代理」，整个模块的准确率会直接塌掉。
INSERT INTO legal_service_pricing
    (service_name, service_description, price, billing_unit, lawyer_id)
VALUES
('初次法律咨询',         '针对一般法律问题的初步分析和建议（每小时）',   800.00,  '按小时', 101),
('合同审查（标准）',     '对标准商业合同进行合规性与风险审查（按次）',  2500.00,  '按次',   102),
('民事诉讼一审代理',     '代理普通民事纠纷案件的一审程序（全案）',     30000.00, '按项目', 105),
('律师函出具',           '就特定事务起草并发出正式律师函',             1500.00,  '按次',   103),
('劳动争议仲裁代理',     '代理用人单位或员工参与劳动仲裁程序',        12000.00, '按项目', 107),
('法律顾问服务（月度）', '提供月度日常法律咨询及简单文件审阅',         5000.00,  '按小时', 102),
('数据合规方案辅导',     '帮助企业初步建立数据合规框架',              15000.00, '按项目', 111);
-- 实际业务里还会有股权激励、尽职调查、知识产权维权等几十行，写法同上。

-- 自证数据进来了，别跳过这一步
SELECT COUNT(*) AS 服务条数, MIN(price) AS 最低价, MAX(price) AS 最高价
FROM legal_service_pricing WHERE is_active = TRUE;

# =============================================================================
# wf_law_assistant.yml —— 法律问答助手的对话流骨架（结构示意，不含任何真实凭据）
#
#   开始 ─► 意图识别 ┬─► 法律咨询：查询改写 → 检索(top_k=3,开 Rerank) → 作答 → 回复
#                    ├─► 案件分析：文档提取 → 条件提取 → 检索(top_k=10) → 分析 → 回复
#                    ├─► 合同分析：文档提取 → 检索(top_k=10) → 合同分类 → 分类审核 → 回复
#                    ├─► 资费查询：数据库查询 → 报价计算 → 回复
#                    └─► 其它：兜底回复
#
# 三个结构性决定：
#   1. 分流放最前面 —— 四路的检索参数、提示词、数据来源都不同，合成一条只会互相打架
#   2. 每一路自带回复节点 —— 共用出口时改一路文案会影响其它三路
#   3. 分流必须有兜底分支 —— 没有「其它」，意外输入会让流程断在半路
#
# 下面完整给出「意图识别 + 法律咨询」这一路，其余三路结构同构、差异见注释。
# 提示词全文见 law_prompts.py。
# =============================================================================
app: {name: 法律问答助手, mode: advanced-chat}   # advanced-chat = 对话流，带会话上下文
kind: app
version: 0.1.5

# 复用锚点，省掉四路重复的模型配置
x-llm: &llm {provider: 'TODO-推理模型供应商', name: 'TODO-推理模型名', mode: chat}
x-rr:  &rr  {provider: 'TODO-Rerank-供应商', model: 'TODO-Rerank-模型'}
x-kb:  &kb  ['TODO-法律条文知识库-ID']

workflow:
  features:
    # 案件与合同两路都要收文件，这里不开，那两路永远拿不到内容
    file_upload: {enabled: true, allowed_file_types: [document],
                  allowed_file_extensions: ['.pdf', '.docx', '.txt', '.md']}
    opening_statement: 你好，我是「明德同学」智能助手。可以咨询法律法规、进行案情分析、分析合同，也可以咨询律所的资费。
    retriever_resource: {enabled: true}      # 展示引用来源，排查召回全靠它

  graph:
    nodes:
      - {id: start, data: {title: 开始, type: start, variables: []}}

      # 分类描述写得像真实用户的话，比写成业务目录准得多；温度压到 0 求稳定
      - id: intent
        data:
          title: 意图识别
          type: question-classifier
          query_variable_selector: [start, sys.query]
          model: {<<: *llm, completion_params: {temperature: 0.0}}
          classes:
            - {id: cls_law,      name: 咨询法律条文、法律规定、某种行为是否合法}
            - {id: cls_case,     name: 描述了一段纠纷经历，希望分析案情、判断责任}
            - {id: cls_contract, name: 提供了合同文本，希望审查条款风险}
            - {id: cls_price,    name: 询问律师费、代理费、收费标准、报价}
            - {id: cls_other,    name: 其它与上述业务无关的问题}

      # --- 路线一：法律咨询 ------------------------------------------------
      # 先改写再检索：用户原话里的情绪和铺垫会在向量空间里把方向带偏
      - id: law_rewrite
        data:
          title: 查询改写
          type: llm
          model: *llm
          memory: {enabled: true}            # 接住「那第二种情况呢」
          prompt_template:
            - {role: system, text: 把用户提问改写成更适合检索知识库的问题，只输出改写后的问题}
            - {role: user, text: '{{#sys.query#}}'}

      - id: law_retrieval
        data:
          title: 法律条文检索
          type: knowledge-retrieval
          dataset_ids: *kb
          query_variable_selector: [law_rewrite, text]   # 用改写后的问题去检索
          retrieval_mode: multiple
          # 查条文原文：多返回的每一条都是噪声，所以压小 top_k 并开 Rerank。
          # 案件分析与合同分析两路相反，用 top_k: 10 —— 参照的条文越多越好。
          multiple_retrieval_config: {top_k: 3, reranking_enable: true, reranking_model: *rr}

      - id: law_answer_llm
        data:
          title: 条文作答
          type: llm
          model: *llm
          context: {enabled: true, variable_selector: [law_retrieval, result]}
          memory: {enabled: true}
          prompt_template:
            - {role: system, text: 先给结论，再给法律原文依据；知识库中没有对应条文时明确告知未检索到}
            - {role: user, text: "用户的问题：\n{{#sys.query#}}\n知识库：\n{{#context#}}"}

      # 每一路自带出口，改一路文案不会影响其它三路
      - {id: law_reply, data: {title: 输出法律条文, type: answer,
         answer: "{{#law_answer_llm.text#}}\n\n以上内容来自律所智能大脑，建议联系律所获取专业的咨询建议。"}}

      # --- 路线二：案件分析（结构同上，多两处差异） ------------------------
      # 1) 开头多一个文档提取器：type: document-extractor，
      #    variable_selector: [start, sys.files]，is_array_file: true
      # 2) 检索前多一个 LLM 节点「检索条件提取」：几千字案情直接去检索，
      #    等于把整个案子的意思平均成一个点，先提炼再检索命中率才上得去

      # --- 路线三：合同分析（比案件分析多一次分流） ------------------------
      # 检索之后再接一个 question-classifier，把合同分成劳动 / 技术服务 / 其它三支
      # （classes 写法同上面的 intent 节点），每支各挂一个 LLM 节点：
      # user 提示词完全一致，只有 system 不同 —— 分流分的就是这一段

      # --- 路线四：资费查询（查数据库，不查知识库） ------------------------
      # 价格是结构化事实，改完要立刻生效，灌进知识库只会答出旧价
      - id: price_db
        data:
          title: 运行数据库查询
          type: tool
          desc: 连接必须用只读账号，查询只指向视图，且必须带 LIMIT
          tool_parameters:
            db_properties: 'charset=utf8mb4&collation=utf8mb4_0900_ai_ci'
            query: 'SELECT * FROM v_pricing_for_llm LIMIT 50'

      - id: price_llm
        data:
          title: 资费计算
          type: llm
          model: *llm
          context: {enabled: true, variable_selector: [price_db, text]}   # 上下文来自数据库
          prompt_template:
            - {role: system, text: 结合定价细则给出报价区间；定价表里没有的服务，如实说明需要人工确认}
            - {role: user, text: "用户问题：\n{{#sys.query#}}\n定价：\n{{#context#}}"}

      # --- 兜底：把能力边界讲清楚，别让用户对着空气发问 --------------------
      - {id: other_reply, data: {title: 兜底回复, type: answer,
         answer: 这个问题超出了我的服务范围。我可以帮你查询法律条文、分析案情、审查合同条款，或者介绍律所的收费标准。}}

    # sourceHandle 填的是分类的 id，填错整路走不到，而画布上看不出异常
    edges:
      - {source: start,  sourceHandle: source,       target: intent}
      - {source: intent, sourceHandle: cls_law,      target: law_rewrite}
      - {source: intent, sourceHandle: cls_case,     target: case_doc}
      - {source: intent, sourceHandle: cls_contract, target: contract_doc}
      - {source: intent, sourceHandle: cls_price,    target: price_db}
      - {source: intent, sourceHandle: cls_other,    target: other_reply}
      - {source: law_rewrite,    sourceHandle: source, target: law_retrieval}
      - {source: law_retrieval,  sourceHandle: source, target: law_answer_llm}
      - {source: law_answer_llm, sourceHandle: source, target: law_reply}
      - {source: price_db,       sourceHandle: source, target: price_llm}