7.1 API集成：把应用接进业务系统
7.1 API 集成：把应用接进业务系统
服务化 API是 Dify 应用的正式出货口：每种应用形态有对应端点，鉴权用应用级 Bearer 密钥，响应分阻塞与流式两种模式。集成方只需要密钥、端点、请求体三样东西，就能把小艺的对话能力嵌进任何系统——帮助中心、小程序、内部工单台。
交付三动作的第一个：接线。3.3 节拿到的 API 密钥在这里正式上岗。本节产出一段可直接进生产的服务端封装；7.2 节会给这把钥匙配上管理制度。
先把三样东西凑齐
端点：聊天助手走消息接口（发消息、收回答），工作流走运行接口（一次输入一次输出）。以聊天助手为例，请求打到平台的对话消息路径上。密钥：应用"访问 API"页创建，格式为 app- 开头的应用密钥，一个应用可持多把。请求体：核心字段五个——
inputs 表单变量的值（3.1 的 member_level 就放这里） query 用户本轮的话 user 用户标识：你系统里的用户 ID，用于会话隔离与用量统计 conversation_id 会话标识：首轮不传，平台返回后后续轮带上，实现多轮 response_mode blocking（等完整答案）或 streaming（逐段接收）
鉴权就一行请求头：Authorization: Bearer app-xxxx。注意这个密钥是应用级的——它代表"这个应用的全部能力"，所以只能在你的服务端使用，绝不下发到浏览器或 App 包里。前端与平台之间，永远隔着你自己的后端。
阻塞与流式：两种模式的选择
阻塞模式：发出去，等几秒，一次性拿完整 JSON。实现最简单，适合后台任务、批量处理、对延迟不敏感的内部系统。缺点是长回答时用户盯着空白。
流式模式：服务端逐段推送（SSE 协议，一段一个事件），前端边收边渲染出打字机效果。用户体感好，适合一切面向真人的场景。事件流里除了回答正文，结束事件还带 token 用量、耗时、引用来源等元数据——做用量统计与引用展示就靠它。
两种模式我各给一段封装。先看阻塞版（Python，可直接跑）：
import requests DIFY_BASE = "https://dify.example.com/v1" # 平台 API 地址 API_KEY = "app-xxxxxxxx" # 应用密钥，来自环境变量更稳妥 def ask_blocking(query, user_id, member_level="普通会员", conversation_id=""): """阻塞式一轮对话：返回回答文本与新会话标识""" resp = requests.post( f"{DIFY_BASE}/chat-messages", headers={"Authorization": f"Bearer {API_KEY}"}, json={ "inputs": {"member_level": member_level}, "query": query, "user": user_id, # 你系统的用户 ID "conversation_id": conversation_id, # 首轮传空串 "response_mode": "blocking", }, timeout=30, # 显式超时，永不裸等 ) resp.raise_for_status() data = resp.json() # data 结构：answer（回答）、conversation_id（会话标识， # 存起来下一轮传回，多轮对话的钥匙）、metadata（用量与引用） return data["answer"], data["conversation_id"] # 使用：首轮 answer, conv = ask_blocking("七天无理由退货怎么办理", "u-10086") # 次轮带上 conv，小艺记得上一轮说了什么 answer2, conv = ask_blocking("那运费谁出", "u-10086", conversation_id=conv)
流式版的关键是解析逐段事件（服务端把每个 SSE 事件作为一个 JSON 行推送）：
def ask_streaming(query, user_id, conversation_id=""): """流式对话：逐段产出回答文本，末段带元数据""" resp = requests.post( f"{DIFY_BASE}/chat-messages", headers={"Authorization": f"Bearer {API_KEY}"}, json={"inputs": {}, "query": query, "user": user_id, "conversation_id": conversation_id, "response_mode": "streaming"}, stream=True, timeout=60, ) resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): if not line or not line.startswith("data:"): continue # SSE 里非 data 行直接跳过 payload = json.loads(line[len("data:"):]) if payload.get("event") == "message": yield payload.get("answer", "") # 正文分片，边收边渲染 elif payload.get("event") == "message_end": meta = payload.get("metadata", {}) # 用量与引用在结束事件里 yield f"\n[本次消耗 {meta.get('usage', {}).get('total_tokens')} tokens]" # 前端拿到逐段文本即可渲染打字机效果 for chunk in ask_streaming("我的订单到哪了", "u-10086"): print(chunk, end="", flush=True)
图 7-2：集成拓扑：你的后端是唯一的持钥者
生产化的四件小事
示例之外，进生产还要补四件。其一，密钥进环境变量，代码仓库里只有变量名；泄露的处置是"创建新密钥、替换、删除旧密钥"，平台侧即时生效。其二，超时与重试：只重试幂等的读操作，写类（建工单）请求重试前必须带幂等键防重复建单。其三，限流：按 user 维度在你的后端限流（平台侧也有限流，双重保护），防一个异常客户端刷爆账单。其四，错误分流：400 类（参数错）记日志告警开发，429（限流）退避重试，5xx 熔断降级到静态 FAQ 页——降级预案是可用性的最后一道网。
变式：三种典型集成形态
变式一，轻嵌入：不想动后端时，用 3.3 节的嵌入代码直接挂浮窗——代价是功能与品牌都受限于现成组件。变式二，全接管：如上所述经自有后端中转，可控性最高，帮助中心采用的就是这个。变式三，纯任务型：不走对话，定时任务批量调工作流接口（比如每晚跑一遍待分类留言），注意并发别超过平台限流阈值，用队列串行化最省心。
⚠️ 常见坑一：前端直连平台 API，密钥打进页面代码——等于把应用钥匙复制给每一个访客。坑二：会话标识没持久化，用户刷新页面小艺就失忆，投诉"它记性差"其实是集成层丢了 conversation_id。坑三：流式响应当阻塞解析，一个 chunk 一个 JSON 解析报错满天飞——先读懂事件协议再写解析。
本节要点回顾
- 三样凑齐：端点按形态选、密钥应用级、请求体五字段；
- 模式选择：面向真人一律流式，后台与批量用阻塞；
- 两个标识：user 管"你是谁"，conversation_id 管"聊到哪"，各司其职；
- 后端中转：浏览器永不见密钥，你的后端是唯一持钥者；
- 生产四件：密钥入环境变量、幂等重试、按用户限流、错误分流降级；
- 结束事件有元数据：用量与引用在流式的收尾事件里，统计就靠它。
线接好了，钥匙的保管制度、谁能改小艺、测试与正式怎么隔离——下一节筑墙。