Glean Platform Chat API：构建带权限上下文的企业助手

POST /api/chat 做基于公司知识的助手回合；需要 CHAT scope。流式时在 JSON 里设 stream: true（不是只靠 Accept 头）。

先看清成品

• mock 返回 Grounded answer。
• 1 项 pytest 通过。

合规提醒：Glean 是企业产品，需管理员采购与部署；个人通常不能自助注册。搜索与聊天结果继承各数据源权限，禁止用管理员特权接口绕过 ACL 给无授权用户。Token 等同企业访问权，勿提交公开仓库。本文不编造报价。

运行说明：无公司 Glean 租户。A/B 用本地 mock 演示 Platform Search/Chat 形状；C 为发查询前的脱敏守卫。未连接真实 *-be.glean.com。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

108-glean-B-src/
├── glean_chat.py
├── mock_glean.py
└── test_108b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Glean Platform Chat: POST /api/chat (CHAT scope)."""
from __future__ import annotations
import argparse, json, os, sys, urllib.error, urllib.request

def chat(base, token, message, stream=False):
    payload = {"messages": [{"author": "USER", "fragments": [{"text": message}]}], "stream": stream}
    req = urllib.request.Request(
        f"{base.rstrip('/')}/api/chat",
        data=json.dumps(payload).encode(),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e

def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("message")
    ap.add_argument("--base", default=os.environ.get("GLEAN_BASE_URL", "https://example-be.glean.com"))
    ap.add_argument("--token", default=os.environ.get("GLEAN_TOKEN", ""))
    args = ap.parse_args(argv)
    if not args.token:
        print("设置 GLEAN_TOKEN（需 CHAT scope）", file=sys.stderr); return 2
    print(json.dumps(chat(args.base, args.token, args.message), ensure_ascii=False, indent=2))
    return 0

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

第 2 步：测试

import subprocess, sys, time
from pathlib import Path
HERE = Path(__file__).resolve().parent

def test_chat():
    port = 19181
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_glean.py"), "--port", str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run([sys.executable, str(HERE/"glean_chat.py"), "政策是什么", "--base", f"http://127.0.0.1:{port}", "--token", "glean-test-token"], capture_output=True, text=True, check=True)
        assert "Grounded" in r.stdout
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 3 步：运行

$ pytest glean chat
.                                                                        [100%]
1 passed in 0.36s

怎么检查结果？

• 输出含 messages。
• 正式调用勿把密钥写进 message。

调试示例

只改 Accept: text/event-stream 却不流式。 官方要求 body 里 stream=true。

常见问题

模板规则

• CHAT token 与 SEARCH 分离更好。
• 答案要可追溯。
• 禁止套话绕过合规审查。

项目资源

助手能答，不代表你可以外传——看清权限与密级。