豆包：给网店做一个客服网页：Flask 后端，回答只依据 faq.md，用 SSE 流式输出到页面

给网店做一个客服网页：Flask 后端，回答只依据 faq.md，用 SSE 流式输出到页面。超出知识库的问题（比如“能打八折吗”）引导转人工，并且把转人工的那轮对话从历史里去掉，不让它影响后面的回答。简单问答关闭深度思考（thinking.type=disabled）。

先看清成品

• 测试： pytest 2 项通过。
• Playwright 打开网页实测（本地 qwen2.5 1.5B），三轮对话：
  • 周末下单几时发货：回答周一发货，正确。
  • 能打八折吗：显示转人工提示。
  • 退货运费谁出：回答由买家承担，但漏了“质量问题除外”。
• 截图： 43-doubao-B-src/ui.png。

没有在 box 上跑的部分： box 上没有方舟 API Key，没有调用过豆包 / 火山方舟的真实接口。代码按官方文档的 Chat API（OpenAI 兼容，base_url https://ark.cn-beijing.volces.com/api/v3）编写，测试时把 ARK_BASE_URL 指向两种替身：

• 本地 Ollama。
• 我自己写的模拟服务器：它只照着文档检查了请求格式，返回的内容是固定的假数据。

合规提醒：方舟 API Key（ARK_API_KEY）只放在环境变量里，不要写进网页前端。商品图、用户评论和聊天记录都可能含有个人信息，上传前要确认有权使用、做好脱敏。AI 生成的商品描述和客服回答必须经过人工审核才能发布，不能写夸大或无法证实的功效（《广告法》）。对公众提供生成式 AI 服务，需要按《生成式人工智能服务管理暂行办法》《人工智能生成合成内容标识办法》等规定履行相应义务。价格以方舟控制台的“模型定价”为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

43-doubao-B-src/
├── app.py
├── faq.md
├── test_app.py
└── ui.png

第 1 步：代码

app.py：

"""网店客服小助手（豆包，火山方舟 Chat API）：只根据 faq.md 回答，流式输出到网页；答不了的引导转人工。
简单问答关闭深度思考（thinking.type=disabled），响应更快、tokens 更少。
环境变量：ARK_API_KEY、ARK_BASE_URL（默认 https://ark.cn-beijing.volces.com/api/v3）、ARK_MODEL（默认 doubao-seed-2-1-lite-260915）
运行：python app.py  然后打开 http://127.0.0.1:5000"""
import json, os
from flask import Flask, Response, request
from openai import OpenAI

BASE = os.environ.get("ARK_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
client = OpenAI(base_url=BASE, api_key=os.environ.get("ARK_API_KEY") or "missing-key")
MODEL = os.environ.get("ARK_MODEL", "doubao-seed-2-1-lite-260915")
FAQ = open(os.path.join(os.path.dirname(__file__), "faq.md"), encoding="utf-8").read()
SYSTEM = ("你是网店客服，语气礼貌简洁。只能根据下面的知识库回答；知识库里没有的问题，回答“这个问题我需要请人工客服确认”，"
          "并在回答末尾加上标记 [转人工]。不要承诺知识库之外的优惠、赔偿或时间。\n\n" + FAQ)
app = Flask(__name__)

PAGE = """<!doctype html><meta charset="utf-8"><title>客服</title>
<style>body{font-family:sans-serif;max-width:640px;margin:2em auto}#log div{margin:.5em 0}.u{color:#555}.h{color:#c0392b}</style>
<div id="log"></div><form id="f"><input id="q" style="width:80%" placeholder="请输入问题"><button>发送</button></form>
<script>
const log=document.getElementById('log'), hist=[];
document.getElementById('f').onsubmit=async e=>{e.preventDefault();
  const q=document.getElementById('q').value; if(!q) return; document.getElementById('q').value='';
  log.insertAdjacentHTML('beforeend',`<div class="u">你：${q.replace(/</g,'&lt;')}</div>`);
  const d=document.createElement('div'); d.className='a'; d.textContent='客服：'; log.appendChild(d);
  hist.push({role:'user',content:q});
  const r=await fetch('/chat',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(hist)});
  const rd=r.body.getReader(), dec=new TextDecoder(); let buf='', ans='';
  for(;;){const {done,value}=await rd.read(); if(done) break; buf+=dec.decode(value,{stream:true});
    let i; while((i=buf.indexOf('\\n\\n'))>=0){const line=buf.slice(0,i); buf=buf.slice(i+2);
      if(!line.startsWith('data: ')) continue; const ev=JSON.parse(line.slice(6));
      if(ev.delta){ans+=ev.delta; d.textContent='客服：'+ans.replace('[转人工]','');}
      if(ev.handoff){d.insertAdjacentHTML('beforeend','<span class="h">（已为你转接人工客服，9:00–21:00 在线）</span>');}}}
  hist.push({role:'assistant',content:ans});};
</script>"""

HANDOFF_PHRASE = "需要请人工客服确认"

def is_handoff(text):
    # 模型经常只说“需要人工确认”却忘了加 [转人工] 标记，所以两种都认
    return "[转人工]" in text or HANDOFF_PHRASE in text

def drop_handoff_turns(history):
    """已经转人工的那一问一答不再发给模型：否则小模型会照着上一轮的拒答句式，把能回答的问题也拒掉"""
    out = []
    for m in history:
        if m["role"] == "assistant" and is_handoff(m["content"]):
            if out and out[-1]["role"] == "user": out.pop()
            continue
        out.append(m)
    return out

@app.get("/")
def index(): return PAGE

@app.post("/chat")
def chat():
    history = drop_handoff_turns(request.get_json())[-10:]      # 只带最近 10 条，控制上下文长度
    extra = {} if "11434" in BASE else {"thinking": {"type": "disabled"}}
    def gen():
        full = ""
        for ch in client.chat.completions.create(model=MODEL, stream=True, temperature=0.3, extra_body=extra,
                                                 messages=[{"role": "system", "content": SYSTEM}, *history]):
            if ch.choices and ch.choices[0].delta.content:
                full += ch.choices[0].delta.content
                yield f"data: {json.dumps({'delta': ch.choices[0].delta.content}, ensure_ascii=False)}\n\n"
        yield f"data: {json.dumps({'done': True, 'handoff': is_handoff(full)})}\n\n"
    return Response(gen(), mimetype="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

if __name__ == "__main__":
    app.run(port=int(os.environ.get("PORT", 5000)))

faq.md：

# 示例小店客服知识库（虚构）
- 发货：工作日 16:00 前下单当天发货，周末订单周一发货。默认中通快递。
- 退换：签收 7 天内可无理由退货，商品需保持完好；运费由买家承担，质量问题除外。
- 发票：可开电子普通发票，在订单页“申请开票”填写抬头和税号。
- 营业时间：人工客服 9:00–21:00。

test_app.py：

import os; os.environ.setdefault("ARK_API_KEY", "test")
from app import drop_handoff_turns, is_handoff

def test_handoff_detected_without_marker():
    assert is_handoff("这个问题我需要请人工客服确认。") and is_handoff("好的[转人工]") and not is_handoff("周一发货")

def test_handoff_turns_dropped():
    h = [{"role": "user", "content": "几点发货"}, {"role": "assistant", "content": "16:00 前当天发"},
         {"role": "user", "content": "打八折吗"}, {"role": "assistant", "content": "这个问题我需要请人工客服确认。"},
         {"role": "user", "content": "退货运费谁出"}]
    assert [m["content"] for m in drop_handoff_turns(h)] == ["几点发货", "16:00 前当天发", "退货运费谁出"]

第 2 步：运行

$ export ARK_API_KEY=...
$ python app.py          # 打开 http://127.0.0.1:5000
# box 上：ARK_BASE_URL=http://127.0.0.1:11434/v1 ARK_MODEL=qwen2.5-local:1.5b python app.py

检查点：问“能打八折吗”，页面上应该出现转人工的提示；接着问别的问题，回答不能受刚才那轮的影响。

调试示例

1. 模型没有输出约定的标记。 提示词要求转人工时输出“[转人工]”。小模型经常不输出这个标记，只说“这个问题我需要请人工客服确认”。所以 is_handoff 同时认标记和这句话。

2. 回答漏了条件。 问“退货运费谁出”，模型回答“买家承担”，漏掉了 FAQ 里的“质量问题除外”。这种漏条件的情况程序很难自动发现，处理办法有两个：一是换更强的模型；二是把 FAQ 里带条件的条目写成独立的一句，比如“质量问题退货，运费由店铺承担”。客服的话术上线前要逐条测一遍。

常见问题

模板规则

• 只根据知识库回答，答不了就转人工。
• 转人工的那轮不留在对话历史里。
• 带条件的规则要写成独立的一句。

项目资源

客服机器人最重要的能力不是什么都能答，而是知道哪些问题不该自己答。