用 RAGFlow API 构建可靠的流式对话客户端：SSE 与引用解析

RAGFlow 的对话接口默认流式返回，每条 SSE 事件带一段回答，可能还带引用和“思考”标记。 这篇做 rf_stream.py：正确拼出完整回答，把思考过程和正式回答分开，取出引用，并同时兼容 1.0 的增量格式和 legacy 的累计格式。

先看清成品

• 对本地 mock 真实运行，第一版脚本连续踩到三个真实问题：中文被截断导致 JSON 解析失败；思考过程混进回答；legacy 模式下回答重复了 6 遍。
• 修复后两种模式输出完全相同的回答、思考和引用。
• 6 项 pytest 通过，其中一项复现了截断的原因。

合规提醒：RAGFlow 是 Apache 2.0 开源项目。本文只用官方 HTTP API 和你自己实例的 API Key（在 RAGFlow 网页的 API 设置中创建）。Key 只放服务端。demo.ragflow.io 是官方的公开演示站，我只用无效 Key 对它发了几个请求来确认错误格式，没有注册、上传或压测；不要把它当成自己的生产服务，也不要上传敏感文档。

运行说明：RAGFlow 官方推荐至少 4 核 CPU、16 GB 内存、50 GB 磁盘，并用 Docker 部署。box 没有 Docker 权限，所以没有部署 RAGFlow。真实跑过的是：用无效 Key 调用 demo.ragflow.io（2026-10-21 21:31–21:33，UTC+8，返回原样贴在文中），以及全部本地代码。流式事件来自我写的 mock，字段照官方 HTTP API 参考。版本提醒：2026-10-16 发布了 v1.0.0-rc2，main 分支的 API 文档已经按 1.0（Go 后端）写，有几个旧接口被移除；demo 站的响应没有 1.0 才有的 X-API-Source: go 头。先确认你的版本，再看对应版本的文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

84-ragflow-C-src/
├── rf_stream.py
├── rf_stream_v0.py
├── mock_ragflow_sse.py
└── test_rf_stream.py

第 1 步：本地 mock（我写的）

"""本地假 RAGFlow 流式对话接口（我写的，仅用于演示）。事件结构照官方 HTTP API 参考
“Converse with chat assistant”的流式示例：默认逐段增量（1.0），legacy=true 时每次返回累计全文（v0.23 兼容）。
reference.chunks 里每一项的字段，官方这一节没有给示例，我借用了检索接口 chunk 的字段。"""
import json, time
from http.server import BaseHTTPRequestHandler, HTTPServer

THINK = ["用户问年假，", "去员工手册里找。"]
ANSWER = ["入职满一年后，", "每年有 10 天带薪年假，", "可顺延到次年 3 月 31 日。"]
REF = {"chunks": [{"id": "c1", "content": "入职满一年后，每年享有 10 天带薪年假……", "document_id": "d1",
                   "document_keyword": "员工手册.md", "similarity": 0.81}]}


def events(legacy):
    base = {"audio_binary": None, "prompt": "", "final": False, "id": "m1", "session_id": "s1", "chat_id": "chat1"}
    if legacy:
        acc = "<think>"
        for t in THINK:
            acc += t; yield {**base, "answer": acc, "reference": {"chunks": []}}
        acc += "</think>"
        for t in ANSWER:
            acc += t; yield {**base, "answer": acc, "reference": {"chunks": []}}
        yield {**base, "answer": acc, "reference": REF, "final": True}
        return
    yield {**base, "answer": "", "reference": {"chunks": []}, "start_to_think": True}
    for t in THINK:
        yield {**base, "answer": t, "reference": {"chunks": []}}
    yield {**base, "answer": "", "reference": {"chunks": []}, "end_to_think": True}
    for t in ANSWER:
        yield {**base, "answer": t, "reference": {"chunks": []}}
    yield {**base, "answer": "", "reference": REF, "final": True}


class H(BaseHTTPRequestHandler):
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if self.headers.get("Authorization") != "Bearer ragflow-local-mock":
            b = json.dumps({"code": 401, "data": None, "message": "<Unauthorized '401: Unauthorized'>"}).encode()
            self.send_response(401); self.send_header("Content-Type", "application/json"); self.end_headers(); self.wfile.write(b); return
        self.send_response(200); self.send_header("Content-Type", "text/event-stream"); self.end_headers()
        for e in events(body.get("legacy", False)):
            self.wfile.write(("data:" + json.dumps({"code": 0, "message": "success", "data": e}, ensure_ascii=False) + "\n\n").encode())
            self.wfile.flush(); time.sleep(0.05)
        self.wfile.write(b'data:{"code": 0, "message": "success", "data": true}\n\n')

    def log_message(self, *a):
        pass


if __name__ == "__main__":
    HTTPServer(("127.0.0.1", 9380), H).serve_forever()

第 2 步：第一版（有问题）

# 第一版（有问题，保留做对照）：用 decode_unicode 读行，直接把每个事件的 answer 拼起来
import json, os, sys, requests
r = requests.post(os.environ["RAGFLOW_BASE_URL"] + "/api/v1/chat/completions", stream=True, timeout=120,
                  headers={"Authorization": "Bearer " + os.environ["RAGFLOW_API_KEY"]},
                  json={"chat_id": os.environ["RAGFLOW_CHAT_ID"], "stream": True, "legacy": "--legacy" in sys.argv,
                        "messages": [{"role": "user", "content": sys.argv[1]}]})
text = ""
for line in r.iter_lines(decode_unicode=True):
    if line.startswith("data:"):
        d = json.loads(line[5:])["data"]
        if d is True:
            break
        text += d["answer"]
print(text)

第 3 步：修正后的客户端

"""RAGFlow 流式对话客户端（POST /api/v1/chat/completions，1.0 的统一入口）。

处理三件事：
1. 按字节读 SSE、按 UTF-8 解码（不依赖 requests 猜编码）；
2. 1.0 默认每个事件是增量；legacy=true（兼容 v0.23）时每个事件是累计全文；
3. start_to_think / end_to_think 之间是思考过程，和正式回答分开；legacy 模式里思考在 <think>…</think> 中。
"""
import argparse, json, os, re, sys
import requests

UA = "binbho-tutorial/1.0"


def sse_events(byte_lines):
    for raw in byte_lines:
        line = raw.decode("utf-8").strip() if isinstance(raw, bytes) else raw.strip()
        if not line.startswith("data:"):
            continue
        msg = json.loads(line[5:])
        if msg.get("code") != 0:
            raise RuntimeError(f"code={msg.get('code')} {msg.get('message')}")
        if msg.get("data") is True:      # 官方：流的最后一条是 data: true
            return
        yield msg["data"]


def assemble(events, legacy=False):
    answer, thinking, refs, in_think = "", "", [], False
    for e in events:
        if e.get("start_to_think"):
            in_think = True
        piece = e.get("answer") or ""
        if legacy:
            answer = piece                     # 累计全文：直接覆盖
        elif in_think:
            thinking += piece
        else:
            answer += piece
        if e.get("end_to_think"):
            in_think = False
        chunks = (e.get("reference") or {}).get("chunks") or []
        if chunks:
            refs = chunks                      # 官方：不是每条都带引用，取最后一次非空的
    if legacy:
        m = re.match(r"(?s)\s*<think>(.*?)</think>(.*)", answer)
        if m:
            thinking, answer = m.group(1), m.group(2)
    return answer.strip(), thinking.strip(), refs


def ask(base, key, chat_id, question, session_id=None, legacy=False):
    body = {"chat_id": chat_id, "stream": True, "legacy": legacy, "messages": [{"role": "user", "content": question}]}
    if session_id:
        body["session_id"] = session_id
    r = requests.post(base.rstrip("/") + "/api/v1/chat/completions", json=body, stream=True, timeout=300,
                      headers={"Authorization": f"Bearer {key.strip()}", "User-Agent": UA})
    if "application/json" in r.headers.get("Content-Type", ""):   # 出错时不是 SSE
        d = r.json()
        raise RuntimeError(f"HTTP {r.status_code} code={d.get('code')} {d.get('message')}")
    return assemble(sse_events(r.iter_lines()), legacy)


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("question"); ap.add_argument("--session"); ap.add_argument("--legacy", action="store_true")
    ap.add_argument("--show-thinking", action="store_true")
    a = ap.parse_args(argv)
    try:
        ans, think, refs = ask(os.environ.get("RAGFLOW_BASE_URL", "http://localhost"), os.environ.get("RAGFLOW_API_KEY", ""),
                               os.environ["RAGFLOW_CHAT_ID"], a.question, a.session, a.legacy)
    except (RuntimeError, requests.RequestException) as e:
        print(f"对话失败：{e}", file=sys.stderr); return 1
    if a.show_thinking and think:
        print(f"（思考）{think}\n")
    print(ans)
    print(f"\n引用 {len(refs)} 条：" + "；".join(str(c.get("document_keyword") or c.get("document_name") or c.get("id")) for c in refs))
    return 0


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

第 4 步：测试

import json
import pytest
import mock_ragflow_sse as m
import rf_stream as s


def lines(legacy):
    out = [("data:" + json.dumps({"code": 0, "message": "success", "data": e}, ensure_ascii=False)).encode() for e in m.events(legacy)]
    return out + [b"", b'data:{"code": 0, "message": "success", "data": true}', b'data:{"code": 0, "data": {"answer": "after end"}}']


@pytest.mark.parametrize("legacy", [False, True])
def test_both_modes_same_result(legacy):
    ans, think, refs = s.assemble(s.sse_events(lines(legacy)), legacy)
    assert ans == "入职满一年后，每年有 10 天带薪年假，可顺延到次年 3 月 31 日。"
    assert think == "用户问年假，去员工手册里找。"
    assert refs[0]["document_keyword"] == "员工手册.md"


def test_stops_at_data_true():
    ans, _, _ = s.assemble(s.sse_events(lines(False)))
    assert "after end" not in ans


def test_utf8_bytes_with_nel_byte():
    # “入”的 UTF-8 是 e5 85 a5；按 latin-1 解码后 0x85 变成 NEL，被 splitlines() 当成换行
    raw = 'data:{"code":0,"data":{"answer":"入职满一年"}}'.encode()
    assert list(s.sse_events([raw]))[0]["answer"] == "入职满一年"
    assert len(raw.decode("latin-1").splitlines()) > 1      # 复现第一版的问题


def test_error_event():
    with pytest.raises(RuntimeError, match="code=102"):
        list(s.sse_events([b'data:{"code":102,"message":"bad"}']))


def test_reference_keeps_last_non_empty():
    ev = [{"answer": "a", "reference": {"chunks": [{"id": 1}]}}, {"answer": "b", "reference": {"chunks": []}}]
    assert s.assemble(ev)[2] == [{"id": 1}]

第 5 步：运行

$ python mock_ragflow_sse.py &     # 监听 127.0.0.1:9380
$ export RAGFLOW_BASE_URL=http://127.0.0.1:9380 RAGFLOW_API_KEY=ragflow-local-mock RAGFLOW_CHAT_ID=chat1
21:34 UTC+8
$ python rf_stream.py "年假有几天？" --show-thinking
（思考）用户问年假，去员工手册里找。

入职满一年后，每年有 10 天带薪年假，可顺延到次年 3 月 31 日。

引用 1 条：员工手册.md
$ python rf_stream.py "年假有几天？" --legacy --show-thinking
（思考）用户问年假，去员工手册里找。

入职满一年后，每年有 10 天带薪年假，可顺延到次年 3 月 31 日。

引用 1 条：员工手册.md
$ RAGFLOW_API_KEY=wrong python rf_stream.py "年假有几天？"
对话失败：HTTP 401 code=401 <Unauthorized '401: Unauthorized'>
$ python -m pytest -q
6 passed

怎么检查结果？

• 换成真实实例后，先用 --show-thinking 看一次，确认思考和回答分得对。
• 引用数为 0 的回答，用 A 篇检查是不是没检索到。
• 升级 RAGFlow 前后各跑一次，对比输出格式有没有变化。

调试示例

1. 中文把一行 JSON 截断了。 第一版运行就报：

json.decoder.JSONDecodeError: Unterminated string starting at: line 1 column 158 (char 157)

把出错的行打印出来，是 'data:{"code": 0, …, "answer": "å'，同时 r.encoding 是 ISO-8859-1。原因：mock 返回的 Content-Type: text/event-stream 没写 charset，requests 对 text/* 默认按 ISO-8859-1 解码。“入”的 UTF-8 是 e5 85 a5，按 ISO-8859-1 解码后 0x85 变成 NEL 控制字符，iter_lines(decode_unicode=True) 内部用 splitlines() 把它当成了换行，一行 JSON 被切成两半。修复：按字节读行，自己用 UTF-8 解码。真实服务端如果写了 charset=utf-8 就不会触发，但客户端不应该依赖这一点。

2. 思考过程混进回答，legacy 模式重复。 修好编码后，第一版的两次输出是：

用户问年假，去员工手册里找。入职满一年后，每年有 10 天带薪年假，可顺延到次年 3 月 31 日。
<think>用户问年假，<think>用户问年假，去员工手册里找。<think>用户问年假，去员工手册里找。</think>入职满一年后，<think>用户问年假，…（同一段内容又重复了几遍）

1.0 格式里，思考内容也放在 answer 字段，靠 start_to_think/end_to_think 区分；legacy 格式每条是累计全文，直接拼接会重复。assemble() 按模式分别处理：增量模式按标记分流，legacy 模式用最后一条覆盖，再拆出 <think>。

常见问题

模板规则

• SSE 按字节读，自己解码。
• 先确认是增量还是累计格式。
• 思考和回答分开存。
• 看到 data: true 就停。

项目资源

流式输出看起来简单，编码和格式差一点，用户看到的就是乱码或重复。