用 FastGPT OpenAPI 调用知识库问答：解析回答、引用与错误码

FastGPT 是开源的知识库问答和 AI 工作流平台。做好的应用可以通过 OpenAPI 调用，接口格式和 OpenAI 的 chat/completions 基本一致。 这篇做 fastgpt_chat.py：从命令行向应用提问，打印回答和它引用的知识库片段；没有引用时提醒你核对。重点处理 FastGPT 的错误格式：错误码在 body 里，HTTP 状态码不一定可靠。

先看清成品

• 真实调用云服务：无效 Key 返回的是 HTTP 500 加 code: 514 unAuthApiKey，不是 401。脚本据此给出“API Key 无效”的提示。
• detail=true 时从 responseData 里宽松地提取 quoteList，字段变化也不会崩。
• 6 项 pytest 通过（含实测的 514 返回、HTTP 200 但 code≠200、非 JSON 的网关错误）。

合规提醒：只用 FastGPT 官方 OpenAPI 和你自己账号的 API Key。官方文档说明 API Key“相当于当前账号在当前团队下的访问凭证”，所以只能放在服务端，不要写进网页或 App。开源版是 Apache 2.0 加附加条件：可以商用，但做类似 FastGPT 的多租户 SaaS、或去掉控制台的 LOGO/版权信息，需要另外取得商业授权（以仓库 LICENSE 为准）。上传到知识库的文档要有权使用，不放个人隐私。

运行说明：box 不能用 Docker，也没有 FastGPT 账号，所以没有部署或调用真实的 FastGPT 应用。真实跑过的是：用无效 Key 调用 FastGPT 云服务（cloud.fastgpt.cn 和 cloud.fastgpt.io，2026-10-21 21:26–21:28，UTC+8），返回原样贴在文中；以及全部本地代码。正常的回答和引用来自我写的 mock，结构照官方文档 detail=true, stream=false 的响应示例；官方也说明不同版本 responseData 的字段可能不同。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

83-fastgpt-A-src/
├── fastgpt_chat.py
└── test_fastgpt_chat.py

第 1 步：脚本

"""调用 FastGPT 应用的对话接口（OpenAPI /api/v1/chat/completions），打印回答和引用的知识库片段。

FastGPT 的错误不一定用 HTTP 状态码表达：我实测无效 Key 返回 HTTP 500 + {"code":514,"statusText":"unAuthApiKey"}。
所以这里同时检查 HTTP 状态码和 body 里的 code。
"""
import argparse, json, os, sys, uuid
import requests

UA = "binbho-tutorial/1.0"
KNOWN = {514: "API Key 无效（unAuthApiKey）"}


class FastGPTError(Exception):
    pass


def call(base, key, app_id, question, chat_id=None, session=None):
    s = session or requests
    body = {"appId": app_id, "chatId": chat_id, "stream": False, "detail": True,
            "messages": [{"role": "user", "content": question}]}
    if not chat_id:
        body.pop("chatId")          # 不传 chatId：不使用 FastGPT 的会话上下文
    r = s.post(base.rstrip("/") + "/v1/chat/completions", json=body, timeout=120,
               headers={"Authorization": f"Bearer {key.strip()}", "User-Agent": UA})
    try:
        d = r.json()
    except ValueError:
        raise FastGPTError(f"HTTP {r.status_code}，返回的不是 JSON：{r.text[:120]}")
    code = d.get("code")
    if r.status_code >= 400 or (isinstance(code, int) and code != 200):
        hint = KNOWN.get(code, d.get("message"))
        raise FastGPTError(f"HTTP {r.status_code} code={code} {d.get('statusText', '')}：{hint}")
    return d


def answer_of(d):
    return d["choices"][0]["message"]["content"]


def quotes_of(d):
    """responseData 是各节点的运行结果，字段随版本变化（官方文档原话），所以宽松地找 quoteList。"""
    out = []
    for node in d.get("responseData") or []:
        for q in node.get("quoteList") or []:
            text = (q.get("q") or "") + ("\n" + q["a"] if q.get("a") else "")
            out.append({"id": q.get("id"), "source": q.get("sourceName") or q.get("source"), "text": text.strip()})
    return out


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("question")
    ap.add_argument("--chat-id", help="同一个 chat-id 连续提问会带上下文")
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args(argv)
    base = os.environ.get("FASTGPT_BASE_URL", "http://localhost:3000/api")
    key, app = os.environ.get("FASTGPT_API_KEY", ""), os.environ.get("FASTGPT_APP_ID", "")
    if not key.strip() or not app:
        print("需要 FASTGPT_API_KEY 和 FASTGPT_APP_ID", file=sys.stderr); return 1
    try:
        d = call(base, key, app, a.question, a.chat_id)
    except (FastGPTError, requests.RequestException) as e:
        print(f"调用失败：{e}", file=sys.stderr); return 1
    qs = quotes_of(d)
    if a.json:
        print(json.dumps({"answer": answer_of(d), "quotes": qs}, ensure_ascii=False, indent=2)); return 0
    print(answer_of(d))
    print(f"\n—— 引用 {len(qs)} 条 ——")
    for i, q in enumerate(qs, 1):
        print(f"[{i}] {q['source'] or '未知来源'}：{q['text'][:80]}")
    if not qs:
        print("（没有引用知识库：回答可能来自模型自身，需要核对）")
    return 0


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

第 2 步：测试（mock）

"""mock 是我写的，结构按 FastGPT 官方文档“detail=true, stream=false”的响应示例。"""
import pytest
import fastgpt_chat as f


class R:
    def __init__(self, code, body):
        self.status_code, self._b, self.text = code, body, str(body)
    def json(self):
        if isinstance(self._b, str):
            raise ValueError
        return self._b


class S:
    def __init__(self, resp):
        self.resp, self.sent = resp, None
    def post(self, url, json=None, timeout=None, headers=None):
        self.sent = (url, json, headers); return self.resp


OK = {"id": "x", "model": "", "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
      "choices": [{"message": {"role": "assistant", "content": "年假为 10 天。"}, "finish_reason": "stop", "index": 0}],
      "responseData": [
          {"moduleName": "知识库搜索", "quoteList": [{"id": "q1", "q": "员工年假多少天？", "a": "入职满一年后每年 10 天。", "sourceName": "员工手册.md"}]},
          {"moduleName": "AI 对话", "tokens": 303}]}


def test_ok_and_quotes():
    s = S(R(200, OK))
    d = f.call("http://localhost:3000/api/", " fastgpt-k\n", "app1", "年假几天", session=s)
    assert f.answer_of(d) == "年假为 10 天。"
    assert f.quotes_of(d) == [{"id": "q1", "source": "员工手册.md", "text": "员工年假多少天？\n入职满一年后每年 10 天。"}]
    url, body, headers = s.sent
    assert url == "http://localhost:3000/api/v1/chat/completions"
    assert "chatId" not in body and body["detail"] is True and headers["Authorization"] == "Bearer fastgpt-k"


def test_chat_id_kept():
    s = S(R(200, OK)); f.call("http://h/api", "k", "a", "q", chat_id="c1", session=s)
    assert s.sent[1]["chatId"] == "c1"


def test_invalid_key_real_shape():   # 2026-10-21 实测的返回
    s = S(R(500, {"code": 514, "statusText": "unAuthApiKey", "message": "common:code_error.error_message.514", "data": None}))
    with pytest.raises(f.FastGPTError, match="API Key 无效"):
        f.call("http://h/api", "bad", "a", "q", session=s)


def test_error_code_with_http_200():
    with pytest.raises(f.FastGPTError, match="code=500"):
        f.call("http://h/api", "k", "a", "q", session=S(R(200, {"code": 500, "statusText": "", "message": "应用不存在"})))


def test_non_json():
    with pytest.raises(f.FastGPTError, match="不是 JSON"):
        f.call("http://h/api", "k", "a", "q", session=S(R(502, "<html>Bad Gateway</html>")))


def test_no_quotes():
    assert f.quotes_of({"choices": OK["choices"]}) == []

第 3 步：运行

# 真实请求（2026-10-21 21:26 UTC+8）
$ curl -s -X POST https://cloud.fastgpt.cn/api/v1/chat/completions -H 'Authorization: Bearer fastgpt-invalid-demo' -H 'Content-Type: application/json' -d '{"chatId":"t1","stream":false,"detail":false,"messages":[{"role":"user","content":"你好"}]}'
{"code":514,"statusText":"unAuthApiKey","message":"common:code_error.error_message.514","data":null}   ← HTTP 500
# cloud.fastgpt.io 返回完全相同

$ export FASTGPT_BASE_URL=https://cloud.fastgpt.cn/api FASTGPT_API_KEY=fastgpt-invalid-demo FASTGPT_APP_ID=000000000000000000000000
$ python fastgpt_chat.py "员工年假有几天？"
调用失败：HTTP 500 code=514 unAuthApiKey：API Key 无效（unAuthApiKey）
$ echo $?
1
$ python -m pytest -q
6 passed

# 有真实 Key 时的输出格式（来自 mock，不是真实结果）：
年假为 10 天。

—— 引用 1 条 ——
[1] 员工手册.md：员工年假多少天？
入职满一年后每年 10 天。

怎么检查结果？

• 每个回答都看“引用”：没有引用，说明回答不是来自知识库，要核对。
• 同一个 --chat-id 连续问两次，第二次的“它”能指代上一题，说明上下文生效了。
• 在 FastGPT 的对话日志里能看到 API 调用的记录。

调试示例

按 HTTP 状态码判断错误会被误导。 我原本打算照常用 r.raise_for_status()，所以先用 curl 看了真实的错误返回（见运行输出）：Key 无效时 HTTP 状态是 500，body 是 {"code":514,"statusText":"unAuthApiKey","message":"common:code_error.error_message.514"}。

如果只看状态码，会报 500 Server Error，让人以为是 FastGPT 服务挂了，而不是 Key 写错。message 字段也只是一个翻译键，不是给人看的句子。所以脚本改成：

1. 先解析 JSON，同时看 HTTP 状态码和 body 里的 code（≠200 就算失败）；
2. 用 KNOWN 把已经确认的错误码翻译成中文提示（目前只放了我实测过的 514）；
3. 返回不是 JSON（例如反向代理的 502 页面）时单独提示。

常见问题

模板规则

• Key 只放服务端环境变量。
• 同时检查 HTTP 状态和 body code。
• 没有引用的回答要标记出来。
• chatId 由自己的系统生成，保证唯一。

项目资源

先用命令行把问答和引用跑通，再接进你的服务端。