常见任务

按场景抄代码。每段都是完整可运行的，改掉 BASE 就能用。

所有示例默认服务在 http://127.0.0.1:8864。示例里的 --noproxy '*' 是为了绕开本机代理直连内网服务；如果你的环境没有配代理，去掉也可以。

问一个问题，拿到条文

最常用的一条。CG_RAG 负责检索、重排、让模型选条文，你只拿结果。

curlPythonJavaScript

curl --noproxy '*' -sS \
  -X POST http://127.0.0.1:8864/cg-rag/rag \
  -H 'Content-Type: application/json' \
  -d '{
    "query": "脚手架搭设完毕后，验收应当由谁组织？",
    "scope": "usual",
    "topk": 5,
    "max_items": 3
  }'

import requests

BASE = "http://127.0.0.1:8864"

resp = requests.post(
    f"{BASE}/cg-rag/rag",
    json={
        "query": "脚手架搭设完毕后，验收应当由谁组织？",
        "scope": "usual",
        "topk": 5,
        "max_items": 3,
    },
    timeout=180,
)
resp.raise_for_status()
body = resp.json()

generation = body["generation"]
if not generation.get("ok"):
    raise RuntimeError(generation.get("error"))

print(generation["answer_text"])
for item in generation["pred_items"]:
    print("-", item)

const BASE = "http://127.0.0.1:8864";

const resp = await fetch(`${BASE}/cg-rag/rag`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    query: "脚手架搭设完毕后，验收应当由谁组织？",
    scope: "usual",
    topk: 5,
    max_items: 3,
  }),
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

const { generation } = await resp.json();
if (!generation.ok) throw new Error(JSON.stringify(generation.error));

console.log(generation.answer_text);

响应里最重要的三个字段：

• generation.answer_text：拼好的答案文本，可以直接展示。
• generation.pred_indices：模型选中的候选序号，1-based，对应 retrieval.retrieval_docs 的位置。
• generation.pred_items：按序号取出的条文，已格式化成「《规范名》第 X 条：正文」。

先看 generation.ok

HTTP 200 不代表生成成功。检索成功但模型没选出条文时，HTTP 仍是 200，而 generation.ok 为 false、generation.error.code 为 no_stable_article。业务代码要判断 generation.ok，不要只判断 HTTP 状态码。

只要条文，不要模型

不调用 LLM，最快，也最便宜。适合自己做后处理，或者把候选喂给你自己的模型。

curl --noproxy '*' -sS \
  -X POST http://127.0.0.1:8864/cg-rag/retrieve-rerank \
  -H 'Content-Type: application/json' \
  -d '{"query":"脚手架验收需要检查哪些条文？","scope":"usual","topk":10}'

返回的每条文档长这样：

{
  "id": "17038",
  "law_name": "DB44-1876-2016轮扣式钢管脚手架安全技术规程",
  "article_number": "9.3.5",
  "contents": "《DB44-1876-2016轮扣式钢管脚手架安全技术规程》 模板支撑架搭设完毕后,应组织相关人员验收,验收合格后方可进入下一工序施工。",
  "text": "模板支撑架搭设完毕后,应组织相关人员验收,验收合格后方可进入下一工序施工。",
  "category": "standard_dibiao",
  "source": "bm25",
  "retrieval_score": 0.9409563541412354
}

source 说明这条是被哪一路召回的（bm25 或稠密检索）。retrieval_score 是重排后的分数，已按降序排列。

同一个 query + scope + topk 在 TTL 内会命中缓存，响应里的 cache.hit 会是 true：

{ "cache": { "hit": false, "entries": 6, "hits": 7, "misses": 16, "ttl_seconds": 300.0 } }

我已经有候选条文，只要模型挑

如果你自己做了检索（或者想复用上一步的结果），可以只调用受约束生成。

import requests

BASE = "http://127.0.0.1:8864"
query = "脚手架搭设完毕后，验收应当由谁组织？"

# 1. 先检索
docs = requests.post(
    f"{BASE}/cg-rag/retrieve-rerank",
    json={"query": query, "scope": "usual", "topk": 10},
    timeout=60,
).json()["retrieval_docs"]

# 2. 再让模型从这批候选里挑
result = requests.post(
    f"{BASE}/cg-rag/constrained-generate",
    json={"query": query, "retrieval_docs": docs, "max_items": 3},
    timeout=180,
).json()

print(result["answer_text"])
print("选中序号：", result["pred_indices"])

retrieval_docs 里的每一项至少要有 law_name、article_number、contents，其它字段会被忽略。超过服务端 max_context_docs（默认 78）的部分会被截断。

边生成边显示（SSE）

/cg-rag/rag/stream 用 Server-Sent Events 返回过程事件，适合前端做动态吐字。

import json
import requests

BASE = "http://127.0.0.1:8864"

with requests.post(
    f"{BASE}/cg-rag/rag/stream",
    json={"query": "脚手架搭设完毕后，验收应当由谁组织？", "scope": "usual", "topk": 5},
    headers={"Accept": "text/event-stream"},
    stream=True,
    timeout=180,
) as resp:
    resp.raise_for_status()
    event = None
    for line in resp.iter_lines(decode_unicode=True):
        if not line:
            continue
        if line.startswith("event:"):
            event = line[6:].strip()
        elif line.startswith("data:"):
            payload = json.loads(line[5:].strip())
            if event == "token":
                print(payload.get("text", ""), end="", flush=True)
            elif event == "final":
                print("\n完成：", payload["generation"]["ok"])
            elif event == "error":
                print("\n出错：", payload)

事件顺序是 progress → retrieval → token* → final；出错时是 error 后紧跟一个 final。完整语义见 SSE、取消与错误。

受约束生成的「流式」是重放，不是真流

受约束生成的答案是服务端按模型选中的序号拼出来的，不是模型逐字写出来的。服务端会在拼好之后把它切成几段 token 事件、按 stream_answer_token_delay_seconds（默认 0.025 秒）依次发出，这样前端能拿到打字机效果，渲染路径也和其它模式一致。

但要清楚：这不会让答案更早出现。第一个 token 到达时，生成其实已经结束了。真正边生成边到达的是视觉接口和自由生成。

分析一张工地照片

图片用 base64 传，不走 multipart。

import base64
import requests

BASE = "http://127.0.0.1:8864"

with open("site.jpg", "rb") as fh:
    encoded = base64.b64encode(fh.read()).decode("ascii")

payload = {
    "question": "请检测图中的安全隐患",
    "images": [{"data": encoded, "mime_type": "image/jpeg"}],
}

# 自由生成：只看图，快，不给法规引用
fast = requests.post(f"{BASE}/cg-rag/vision/answer", json=payload, timeout=180).json()
print(fast["answer"])

# 智能生成：模型自己去检索法规，答案里带 [cite^n] 引用
deep = requests.post(
    f"{BASE}/cg-rag/vision/agentic",
    json={**payload, "scope": "full", "topk": 8},
    timeout=600,
).json()
print(deep["answer"])
for citation in deep.get("citations", []):
    print(citation)

两种模式的取舍：

	自由生成 vision/answer	智能生成 vision/agentic
是否检索法规	否	是，模型自行决定检索几轮
答案里有无引用	无	有 [cite^n]，可追回条文
典型耗时	数秒	一到三分钟
适合	现场快速过一遍	出具有依据的判定

智能生成慢是因为它真的在多轮检索。如果你的场景要的是「秒回」，用自由生成。

字段全集、隐患定级口径见 视觉隐患识别。

在 Agent 里当工具用（MCP）

CG_RAG 通过 Streamable HTTP 暴露 MCP，地址是 http://127.0.0.1:8864/cg-rag/mcp/（结尾的斜杠不能省）。

Claude Code：

claude mcp add --transport http cg-rag http://127.0.0.1:8864/cg-rag/mcp/

其它客户端的通用配置：

{
  "mcpServers": {
    "cg-rag": {
      "type": "http",
      "url": "http://127.0.0.1:8864/cg-rag/mcp/"
    }
  }
}

可用工具：cg_health、cg_list_profiles、cg_retrieve_rerank、cg_constrained_generate、cg_rag、cg_vision_answer、cg_vision_observe、cg_vision_agentic。参数与 HTTP 接口一一对应，详见 MCP 工具。

让 CG_RAG 开机就热

第一次请求某个 scope 时要加载索引，会明显变慢。上线后先 warmup：

curl --noproxy '*' -sS \
  -X POST http://127.0.0.1:8864/cg-rag/warmup \
  -H 'Content-Type: application/json' \
  -d '{"scopes":["usual","full"]}'

注意 retriever 同时只保留一个 active profile，warmup 会顺序加载，最终停在最后一个 scope 上。所以把你最常用的放在数组最后。

超时该设多少

按接口分开设，不要一刀切：

接口	建议客户端超时
/cg-rag/health、/cg-rag/profiles	10 s
/cg-rag/retrieve-rerank	60 s（首次加载索引可能久）
/cg-rag/constrained-generate、/cg-rag/rag	180 s
/cg-rag/vision/answer	180 s
/cg-rag/vision/agentic	600 s，它会多轮检索

服务端自身的上游超时由 CG_RAG_TIMEOUT_SECONDS 控制，默认 120 秒（部署时常调到 180）。客户端超时应该不小于它，否则你会先于服务端断开，日志里只留下一个断连。当前生效值可以在 health 的 generation.timeout_seconds 里看到。