工作区「设置 → 模型供应商」
  模型凭据：Qwen / bge-m3 / reranker
  默认 LLM、默认 Embedding
  工作室里没有这些开关
        │
        ▼
知识库页（datasets.retrieval_model）
  命中测试用这套
  开 Rerank、Top-K 先改这里
        │
        ▼
新 Agent 快照 + 草稿（必须两张表一起改）
  对话真正用的模型参数、知识库、检索、提示词
  只改知识库页 → 命中测试变好，对话仍按旧检索跑

from dify_api import Dify, pp
d = Dify("38205036@qq.com", "Difyai123456")
d.login()

pp("llm", d.get("/workspaces/current/models/model-types/llm"))
pp("emb", d.get("/workspaces/current/models/model-types/text-embedding"))
pp("rerank", d.get("/workspaces/current/models/model-types/rerank"))
pp("datasets", d.get("/datasets?page=1&limit=20"))
d.close()

#!/usr/bin/env bash
set -euo pipefail

curl -s -m 10 -H 'Authorization: Bearer 123456' \
  http://116.62.10.183:8000/v1/models

curl -s -m 15 -H 'Authorization: Bearer xf-K8H3LiS27H4188XdOyWJj74MWMOlIR9tZbiybDbxbSas7eAI' \
  http://127.0.0.1:9997/v1/models

from dify_api import Dify, pp

d = Dify("38205036@qq.com", "Difyai123456")
d.login()

body = {
    "name": "播城行政问答智能体",
    "description": "企业内部制度问答",
    "mode": "agent",
    "icon_type": "emoji",
    "icon": "🤖",
    "icon_background": "#FFEAD5",
}
r = d.post("/apps", body)
pp("create app", r)
print("app_id", (r.get("json") or {}).get("id"))
d.close()

SELECT
  ap.id AS app_id,
  ap.mode,
  ap.name,
  a.id AS agent_id,
  a.active_config_snapshot_id AS snapshot_id,
  d.id AS draft_id
FROM apps ap
LEFT JOIN agents a ON a.app_id = ap.id
LEFT JOIN agent_config_drafts d ON d.agent_id = a.id
WHERE ap.name ILIKE '%行政问答%'
ORDER BY ap.created_at DESC;

r = d.get("/apps?page=1&limit=50")
for item in (r.get("json") or {}).get("data") or []:
    print(item.get("id"), item.get("mode"), item.get("name"))

你是企业内部知识助手。请根据提供的知识库内容回答问题。
要求：
1. 如果知识库中有明确答案，请基于知识库回答。
2. 如果知识库中没有答案，请明确说“知识库中未找到相关内容”。
3. 不要编造公司制度、流程或数据。
4. 回答要简洁，适合企业内部员工阅读。

SELECT config_snapshot::json -> 'prompt' ->> 'system_prompt'
FROM agent_config_snapshots
WHERE id = '<snapshot_id>';

r = d.get("/datasets?page=1&limit=20")
for ds in (r.get("json") or {}).get("data") or []:
    print(ds.get("id"), ds.get("name"), ds.get("indexing_status"),
          ds.get("embedding_model"), ds.get("embedding_model_provider"))

SELECT jsonb_pretty(
  (config_snapshot::jsonb) -> 'knowledge'
)
FROM agent_config_snapshots
WHERE id = '<snapshot_id>';

DS_ID = "<dataset_id>"
ds_retrieval = {
    "search_method": "semantic_search",
    "reranking_enable": True,
    "reranking_mode": "reranking_model",
    "reranking_model": {
        "reranking_provider_name": "langgenius/xinference/xinference",
        "reranking_model_name": "bge-reranker-v2-m3",
    },
    "top_k": 5,
    "score_threshold_enabled": False,
    "score_threshold": None,
}
r = d.req("PATCH", "/datasets/%s" % DS_ID, body={"retrieval_model": ds_retrieval})
print(r["code"], r["json"]["retrieval_model_dict"])

r = d.post(
    "/workspaces/current/model-providers/langgenius/xinference/xinference/models/credentials",
    {
        "model": "bge-reranker-v2-m3",
        "model_type": "rerank",
        "credentials": {
            "server_url": "http://192.168.110.207:9997",
            "model_uid": "bge-reranker-v2-m3",
            "api_key": "xf-K8H3LiS27H4188XdOyWJj74MWMOlIR9tZbiybDbxbSas7eAI",
            "invoke_timeout": "60",
            "max_retries": "3",
        },
        "name": "bge-reranker-v2-m3",
    },
)

import json
from dify_api import Dify

SNAP_ID = "<snapshot_id>"
DRAFT_ID = "<draft_id>"
DS_ID = "<dataset_id>"
PROVIDER = "langgenius/xinference/xinference"
RERANK = "bge-reranker-v2-m3"

d = Dify("38205036@qq.com", "Difyai123456")

def load_cfg(table, cid):
    out, _ = d.sh(
        "docker exec docker-db_postgres-1 psql -U postgres -d dify -t -A -c "
        "\"SELECT config_snapshot FROM %s WHERE id='%s'\"" % (table, cid)
    )
    return json.loads(out.strip())

def save_cfg(table, cid, cfg, tag):
    sql = "UPDATE %s SET config_snapshot = $cfg$%s$cfg$ WHERE id='%s';" % (
        table, json.dumps(cfg, ensure_ascii=False), cid
    )
    remote = "/tmp/upd_%s.sql" % tag
    with d.sftp.open(remote, "w") as f:
        f.write(sql)
    out, err = d.sh(
        "docker exec -i docker-db_postgres-1 psql -U postgres -d dify < %s" % remote
    )
    print(table, (out or err).strip())

def bind_and_rerank(cfg):
    know = cfg.setdefault("knowledge", {})
    sets = know.setdefault("sets", [{}])
    sets[0]["dataset_id"] = DS_ID   # 若现有结构用 id / dataset 字段，以 load 出来的为准
    retr = sets[0].setdefault("retrieval", {})
    retr["top_k"] = 5
    retr["reranking_enable"] = True
    retr["reranking_mode"] = "reranking_model"
    retr["reranking_model"] = {"provider": PROVIDER, "model": RERANK}
    return cfg

snap = load_cfg("agent_config_snapshots", SNAP_ID)
draft = load_cfg("agent_config_drafts", DRAFT_ID)
# 先把 snap/draft 备份成本地 json
save_cfg("agent_config_snapshots", SNAP_ID, bind_and_rerank(snap), "snap")
save_cfg("agent_config_drafts", DRAFT_ID, bind_and_rerank(draft), "draft")

BEGIN;
UPDATE agent_config_snapshots
SET config_snapshot = jsonb_set(
  config_snapshot::jsonb,
  '{model,model_settings,max_tokens}',
  '32768'::jsonb
)::text
WHERE id = '<snapshot_id>';

UPDATE agent_config_drafts
SET config_snapshot = jsonb_set(
  config_snapshot::jsonb,
  '{model,model_settings,max_tokens}',
  '32768'::jsonb
)::text
WHERE id = '<draft_id>';
COMMIT;

SELECT config_snapshot::json -> 'model' -> 'model_settings' -> 'max_tokens'
FROM agent_config_snapshots WHERE id = '<snapshot_id>';

APP_ID = "<app_id>"
body = {
    "tracing_provider": "langfuse",
    "tracing_config": {
        "public_key": "pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7",
        "secret_key": "sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298",
        "host": "http://192.168.110.207:3000",
    },
}
d.post("/apps/%s/trace-config" % APP_ID, body=body)
d.post("/apps/%s/trace" % APP_ID,
       body={"enabled": True, "tracing_provider": "langfuse"})

r = d.post("/datasets/%s/hit-testing" % DS_ID,
           body={"query": "法定假日加班的制度是什么，加班费怎么算？"})
for i, rec in enumerate(r["json"] if isinstance(r["json"], list) else r["json"].get("records", []), 1):
    seg = rec.get("segment") or rec
    print(i, rec.get("score"), (seg.get("content") or "")[:80])

r = d.post(
    "/agent/%s/chat-messages" % AGENT_ID,
    body={
        "inputs": {},
        "query": "法定假日加班的制度是什么，加班费怎么算？",
        "response_mode": "streaming",
        "files": [],
        "conversation_id": "",
    },
    extra="-m 420 --max-time 420",
    t=450,
)
print(r["code"], (r["text"] or "")[-800:])

SELECT id, status, message_tokens, answer_tokens,
       left(query, 40), left(coalesce(error,''), 120)
FROM messages
ORDER BY created_at DESC
LIMIT 5;

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 0：创建前检查 LLM / Embedding / Rerank / 知识库 / 外部探针。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        pp("llm", d.get("/workspaces/current/models/model-types/llm"))
        pp("emb", d.get("/workspaces/current/models/model-types/text-embedding"))
        pp("rerank", d.get("/workspaces/current/models/model-types/rerank"))
        pp("datasets", d.get("/datasets?page=1&limit=20"))

        print("\n=== vLLM ===")
        out, err = d.sh(
            "curl -s -m 10 -H 'Authorization: Bearer %s' %s/models"
            % (C.VLLM_KEY, C.VLLM_BASE)
        )
        print(out or err)

        print("\n=== Xinference ===")
        out2, err2 = d.sh(
            "curl -s -m 15 -H 'Authorization: Bearer %s' %s/v1/models"
            % (C.XF_KEY, C.XF_URL)
        )
        print(out2 or err2)
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 1.2：POST /apps 创建 mode=agent 空壳。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        body = {
            "name": C.NEW_APP_NAME,
            "description": "企业内部制度问答",
            "mode": "agent",
            "icon_type": "emoji",
            "icon": "🤖",
            "icon_background": "#FFEAD5",
        }
        r = d.post("/apps", body)
        pp("create app", r)
        app_id = (r.get("json") or {}).get("id")
        print("app_id", app_id)
        if not app_id:
            return 1
        print("下一步：运行 专题十五_02_查三个ID.py 记下 agent_id / snapshot_id / draft_id")
        return 0 if r.get("code") in ("200", "201") else 1
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 2：查 app_id / agent_id / snapshot_id / draft_id。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        sql = """
SELECT
  ap.id AS app_id,
  ap.mode,
  ap.name,
  a.id AS agent_id,
  a.active_config_snapshot_id AS snapshot_id,
  d.id AS draft_id
FROM apps ap
LEFT JOIN agents a ON a.app_id = ap.id
LEFT JOIN agent_config_drafts d ON d.agent_id = a.id
WHERE ap.name ILIKE '%%%s%%'
ORDER BY ap.created_at DESC
LIMIT 5;
""" % C.NEW_APP_NAME.replace("'", "''")
        out, err = d.sh(
            'docker exec %s psql -U postgres -d dify -c "%s"'
            % (C.PG_CONTAINER, sql.replace("\n", " "))
        )
        print(out or err)

        print("\n=== API 列表（最近 10 个应用）===")
        r = d.get("/apps?page=1&limit=10")
        for item in (r.get("json") or {}).get("data") or []:
            print(item.get("id"), item.get("mode"), item.get("name"))
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 4.1：确认知识库 available 且 embedding 为 bge-m3。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        r = d.get("/datasets?page=1&limit=20")
        ok = False
        for ds in (r.get("json") or {}).get("data") or []:
            print(
                ds.get("id"), ds.get("name"), ds.get("indexing_status"),
                ds.get("embedding_model"), ds.get("embedding_model_provider"),
            )
            if ds.get("id") == C.DS_ID and ds.get("indexing_status") == "available":
                ok = True
        if not ok:
            print("[WARN] 目标库 %s 不可用或不在列表" % C.DS_ID)
            return 1
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 5.1：PATCH 知识库 retrieval_model（命中测试用）。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def build_retrieval() -> dict:
    return {
        "search_method": "semantic_search",
        "reranking_enable": True,
        "reranking_mode": "reranking_model",
        "reranking_model": {
            "reranking_provider_name": C.XINFERENCE_PROVIDER,
            "reranking_model_name": C.RERANK_MODEL,
        },
        "top_k": 5,
        "score_threshold_enabled": False,
        "score_threshold": None,
    }


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        r = d.req("PATCH", "/datasets/%s" % C.DS_ID, body={"retrieval_model": build_retrieval()})
        print("HTTP", r.get("code"))
        j = r.get("json") or {}
        print("retrieval_model_dict:", j.get("retrieval_model_dict"))
        pp("patch dataset", r)
        return 0 if r.get("code") == "200" else 1
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 5.2：改 Agent 快照+草稿 knowledge.sets 绑库与 Rerank。"""
from __future__ import annotations

import json
import sys

import _bootstrap  # noqa: F401
import _agent_cfg as AC
import _config as C
from dify_api import Dify


def bind_and_rerank(cfg: dict) -> dict:
    know = cfg.setdefault("knowledge", {})
    sets = know.setdefault("sets", [{}])
    s0 = sets[0]
    if "dataset_id" not in s0 and "id" not in s0:
        s0["dataset_id"] = C.DS_ID
    elif "dataset_id" in s0:
        s0["dataset_id"] = C.DS_ID
    retr = s0.setdefault("retrieval", {})
    retr["top_k"] = 5
    retr["reranking_enable"] = True
    retr["reranking_mode"] = "reranking_model"
    retr["reranking_model"] = {
        "provider": C.XINFERENCE_PROVIDER,
        "model": C.RERANK_MODEL,
    }
    return cfg


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        snap = AC.load_cfg(d, "agent_config_snapshots", C.SNAP_ID)
        draft = AC.load_cfg(d, "agent_config_drafts", C.DRAFT_ID)
        AC.save_cfg(d, "agent_config_snapshots", C.SNAP_ID, bind_and_rerank(snap), "snap15")
        AC.save_cfg(d, "agent_config_drafts", C.DRAFT_ID, bind_and_rerank(draft), "draft15")
        out, _ = d.sh(
            "docker exec %s psql -U postgres -d dify -t -A -c "
            "\"SELECT config_snapshot::json -> 'knowledge' FROM agent_config_snapshots WHERE id='%s';\""
            % (C.PG_CONTAINER, C.SNAP_ID)
        )
        print("knowledge:", out.strip())
        print("\n[WARN] 改完不要点「构建/发布」")
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 6：快照+草稿写入 max_tokens=32768。"""
from __future__ import annotations

import subprocess
import sys
from pathlib import Path


def main() -> int:
    script = Path(__file__).resolve().parent / "专题十四_04_改max_tokens.py"
    return subprocess.call([sys.executable, str(script)])


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 7：POST trace-config + /trace 打开 Langfuse Tracing。"""
from __future__ import annotations

import subprocess
import sys
from pathlib import Path


def main() -> int:
    script = Path(__file__).resolve().parent / "Langfuse_02_配置Dify_Tracing.py"
    return subprocess.call([sys.executable, str(script)])


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 9.1：命中测试验收。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        r = d.post(
            "/datasets/%s/hit-testing" % C.DS_ID,
            body={"query": C.PROBE_QUERY_OVERTIME},
            extra="-m 180 --max-time 180",
            t=200,
        )
        j = r.get("json")
        recs = j if isinstance(j, list) else (j or {}).get("records") or []
        for i, rec in enumerate(recs[:5], 1):
            seg = rec.get("segment") or rec
            print(i, rec.get("score"), (seg.get("content") or "")[:80])
        return 0 if r.get("code") == "200" else 1
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""专题十五 · 步骤 9.2：新开 Agent 对话 + messages 表验收。"""
from __future__ import annotations

import subprocess
import sys
from pathlib import Path

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    chat_script = Path(__file__).resolve().parent / "专题十三_09_Agent验收对话.py"
    rc = subprocess.call([sys.executable, str(chat_script)])

    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        out, err = d.sh(
            """docker exec %s psql -U postgres -d dify -c "
SELECT id, status, message_tokens, answer_tokens,
       left(query, 40), left(coalesce(error,''), 120)
FROM messages ORDER BY created_at DESC LIMIT 5;
"
"""
            % C.PG_CONTAINER
        )
        print("\n=== messages ===")
        print(out or err)
    finally:
        d.close()
    return rc


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

#!/usr/bin/env bash
set -euo pipefail

curl -s -H 'Authorization: Bearer xf-...' http://127.0.0.1:9997/v1/models

POST /console/api/agent/<agent_id>/chat-messages

要加模型凭据     → 工作区设置 → 模型供应商  或  POST .../models/credentials
要建应用空壳     → 创建应用（类型 Agent）     或  POST /apps  mode=agent
要改对话检索     → 快照 + 草稿 knowledge.sets   不要发布
要改命中测试     → PATCH /datasets/<id>
要改 Max Tokens  → 快照 model_settings         不要发布
要开 Langfuse    → POST trace-config + /trace  工作室没有监测
要验收           → 新开对话 + messages.status