用 Flowise Prediction API 与 sessionId 构建多轮会话客户端

Flowise 搭好 Chatflow 后，业务系统通过 Prediction API 调用。多轮对话需要把 sessionId 或 history 带回去。 这篇做标准库客户端 predict_client.py，以及本地 mock（mock_flowise.py）模拟 200/401/404/400 和按 session 记住名字，方便没有 Flowise 服务时也能练通。

先看清成品

• mock 上真实跑通两轮对话：先说「我叫小明」，再问「我叫什么？」得到「你叫小明。」
• 401 / 404 / 400 均按官方风格返回；4 项 pytest 通过。
• 未连接真实 Flowise 实例。

合规提醒：用自己的 Flowise 实例与模型 API Key，不共享账号；密钥放环境变量或官方凭据管理，不要写进导出的 chatflow JSON。只处理你有权处理的数据；生产环境务必开启登录保护。费用以各模型供应商官网为准，本文不写价格。

运行说明：我没有安装完整的 Flowise 服务（需要 Node.js 全局包或 Docker）。真实运行过的是：导出 JSON 校验器、Prediction 客户端对 mock 服务的请求、.env 检查脚本，以及 pytest（2026-10-23，UTC+8）。Flowise 服务端行为全部来自官方文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

91-flowise-B-src/
├── predict_client.py
├── mock_flowise.py
└── test_91b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Call Flowise Prediction API (official shape) with session memory helpers."""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass, field


@dataclass
class Client:
    base_url: str
    api_key: str
    flow_id: str
    session_id: str = "default"
    history: list[dict] = field(default_factory=list)

    def ask(self, question: str, *, streaming: bool = False) -> dict:
        url = f"{self.base_url.rstrip('/')}/api/v1/prediction/{self.flow_id}"
        payload = {
            "question": question,
            "streaming": streaming,
            "history": list(self.history),
            "overrideConfig": {"sessionId": self.session_id},
        }
        data = json.dumps(payload).encode()
        req = urllib.request.Request(
            url,
            data=data,
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.api_key}",
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                body = json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            detail = e.read().decode(errors="replace")
            raise SystemExit(f"HTTP {e.code}: {detail}") from e
        except urllib.error.URLError as e:
            raise SystemExit(f"连接失败: {e}") from e
        # Keep local history in Flowise roles for the next turn.
        self.history.append({"role": "userMessage", "content": question})
        text = body.get("text") or ""
        self.history.append({"role": "apiMessage", "content": text})
        return body


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("question", nargs="?", default="食堂几点开门？")
    p.add_argument("--base", default=os.environ.get("FLOWISE_BASE_URL", "http://127.0.0.1:3091"))
    p.add_argument("--key", default=os.environ.get("FLOWISE_API_KEY", ""))
    p.add_argument("--flow", default=os.environ.get("FLOWISE_FLOW_ID", "cf-demo-001"))
    p.add_argument("--session", default="demo")
    args = p.parse_args(argv)
    if not args.key:
        print("请设置 FLOWISE_API_KEY（或 --key）", file=sys.stderr)
        return 2
    c = Client(args.base, args.key, args.flow, session_id=args.session)
    r = c.ask(args.question)
    print(json.dumps(r, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
"""Minimal mock of Flowise POST /api/v1/prediction/{id} for local tests."""
from __future__ import annotations

import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse

API_KEY = "fw-test-key"
FLOW_ID = "cf-demo-001"
SESSIONS: dict[str, list[dict]] = {}


class H(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):  # quieter
        pass

    def _auth(self) -> bool:
        auth = self.headers.get("Authorization", "")
        return auth == f"Bearer {API_KEY}"

    def do_POST(self):
        path = urlparse(self.path).path
        if not path.startswith("/api/v1/prediction/"):
            self.send_error(404)
            return
        flow_id = path.rsplit("/", 1)[-1]
        if not self._auth():
            self.send_response(401)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"error":"Unauthorized"}')
            return
        if flow_id != FLOW_ID:
            self.send_response(404)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"error":"Flow not found"}')
            return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        if "question" not in body and "form" not in body:
            self.send_response(400)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"error":"question or form required"}')
            return
        q = body.get("question") or ""
        oc = body.get("overrideConfig") or {}
        sid = oc.get("sessionId") or "anon"
        hist = list(body.get("history") or [])
        SESSIONS.setdefault(sid, []).extend(hist)
        SESSIONS[sid].append({"role": "userMessage", "content": q})
        # Remember last user name mention for a tiny memory demo
        name = None
        for m in SESSIONS[sid]:
            c = m.get("content", "")
            if "我叫" in c and "我叫什么" not in c and "我的名字" not in c:
                name = c.split("我叫", 1)[-1].strip(" 。.!！，,")
                # drop trailing sentence junk after first token-ish chunk
                for sep in ("，", ",", "。", " ", "！", "!"):
                    if sep in name:
                        name = name.split(sep, 1)[0]
                        break
        if "我叫什么" in q or "我的名字" in q:
            text = f"你叫{name}。" if name else "我还不记得你的名字。"
        else:
            text = f"收到：{q}"
        SESSIONS[sid].append({"role": "apiMessage", "content": text})
        resp = {
            "text": text,
            "question": q,
            "chatId": f"chat-{sid}",
            "sessionId": sid,
            "sourceDocuments": [{"pageContent": "食堂开放时间 11:00-13:00", "metadata": {"source": "handbook.md"}}],
        }
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(resp, ensure_ascii=False).encode())


def main():
    import os
    port = int(os.environ.get("PORT", "3091"))
    HTTPServer(("127.0.0.1", port), H).serve_forever()


if __name__ == "__main__":
    main()

第 3 步：测试

import json
import threading
import time
from http.server import HTTPServer

import mock_flowise
from predict_client import Client


def _serve():
    httpd = HTTPServer(("127.0.0.1", 0), mock_flowise.H)
    port = httpd.server_address[1]
    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()
    return httpd, port


def test_ok_and_memory():
    httpd, port = _serve()
    try:
        c = Client(f"http://127.0.0.1:{port}", mock_flowise.API_KEY, mock_flowise.FLOW_ID, session_id="s1")
        r1 = c.ask("你好，我叫小明")
        assert "小明" in r1["text"] or "收到" in r1["text"]
        r2 = c.ask("我叫什么？")
        assert "小明" in r2["text"]
        assert r2["sessionId"] == "s1"
        assert r2.get("sourceDocuments")
    finally:
        httpd.shutdown()


def test_401():
    httpd, port = _serve()
    try:
        c = Client(f"http://127.0.0.1:{port}", "wrong", mock_flowise.FLOW_ID)
        try:
            c.ask("hi")
            assert False, "should exit"
        except SystemExit as e:
            assert "401" in str(e)
    finally:
        httpd.shutdown()


def test_404_flow():
    httpd, port = _serve()
    try:
        c = Client(f"http://127.0.0.1:{port}", mock_flowise.API_KEY, "no-such")
        try:
            c.ask("hi")
            assert False
        except SystemExit as e:
            assert "404" in str(e)
    finally:
        httpd.shutdown()


def test_400():
    # hit mock directly without question
    import urllib.request
    httpd, port = _serve()
    try:
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/v1/prediction/{mock_flowise.FLOW_ID}",
            data=b"{}",
            headers={"Content-Type": "application/json", "Authorization": f"Bearer {mock_flowise.API_KEY}"},
            method="POST",
        )
        try:
            urllib.request.urlopen(req)
            assert False
        except Exception as e:
            assert "400" in str(e) or getattr(e, "code", None) == 400
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8（对 mock）
$ python -c "..."  # 见下方实录
{
  "text": "收到：你好，我叫小明",
  "question": "你好，我叫小明",
  "chatId": "chat-demo",
  "sessionId": "demo",
  "sourceDocuments": [
    {
      "pageContent": "食堂开放时间 11:00-13:00",
      "metadata": {
        "source": "handbook.md"
      }
    }
  ]
}
---
{
  "text": "你叫小明。",
  "question": "我叫什么？",
  "chatId": "chat-demo",
  "sessionId": "demo",
  "sourceDocuments": [
    {
      "pageContent": "食堂开放时间 11:00-13:00",
      "metadata": {
        "source": "handbook.md"
      }
    }
  ]
}

$ python -m pytest -q
4 passed

怎么检查结果？

• 同一 sessionId 下第二轮能复述名字。
• 错误密钥必须 401；错误 flow id 必须 404。
• 有真实 Flowise 时：设 FLOWISE_BASE_URL / FLOWISE_API_KEY / FLOWISE_FLOW_ID 后直接跑客户端。

调试示例

「我叫什么」把名字冲掉了。 mock 第一版用 if "我叫" in content 抽名字，但用户第二句「我叫什么？」也含「我叫」，名字被更新成「什么？」。修复：抽名字时排除「我叫什么」「我的名字」问句。真实 Flowise 里由 LLM+Memory 节点处理，但客户端侧自己维护 history 时同样要注意不要把问句写进“事实”。

常见问题

模板规则

• sessionId 由调用方稳定生成。
• 密钥只放环境变量。
• 本地先对 mock 跑通再接真服务。

项目资源

先让客户端和会话语义正确，再接到真实 Chatflow。