DeepSeek 流式输出与函数调用：把它的能力接进自己程序的细节

Q: 为什么需要 stream

默认情况下，你发一个请求，要等模型把整段回答生成完才一次性返回。复杂问题可能等十几秒，用户还以为卡死了。流式输出让模型边生成边往回发片段（chunk），前端可以像打字机一样逐字显示，体感快很多，也能尽早把「它确实在干活」的信号给用户。

Q: 请求怎么写

在 create 里加 stream=True： from openai import OpenAI client = OpenAI(api_key="你的KEY", base_url="https://api.deepseek.com") stream = client.chat.completions.create( model="deepseek-chat", messages=[{"rol

Q: 底层其实是 SSE

如果你不用 SDK 而想自己解析，流式走的是 Server-Sent Events（SSE）：服务端一行行推 data: {...} 帧，最后一帧是 data: [DONE]。每个帧里的 choices[0].delta.content 是增量文本。理解这点，你自己封装客户端或排查「为什么收不全」时就知道该去解析流式帧，而不是等一个完整 JSON。

Q: 几个容易踩的细节

chunk 可能为空。流刚开始或结束时有控制帧，没有 content，直接 print 会报错，务必加 if delta 判断。stream=True 时拿不到完整对象。想做字数统计、落库，得自己在循环里拼接 full_text += delta。异常发生在流中间。普通请求报错会直接抛异常；流式请求可能前几段正常、中段断流。建议用 try/except 包住整个循环，并在断流时给用户一个「生成中断

Q: 二、函数调用：让模型「动手」

模型本身不能查天气、不能读你的数据库、不能发短信。函数调用就是给它一张「工具清单」，它判断需要时，返回一个调用指令，由你的程序真正去执行，再把结果喂回去。这一步让模型从一个「嘴强王者」变成能操作真实系统的 Agent。

前面几篇都在聊怎么「用」DeepSeek，这篇阿文带大家下沉到工程层：怎么把它的能力嵌进你自己的程序。两个最常被用到的能力——流式输出（stream） 和 函数调用（function calling）。前者决定用户体验的「打字机效果」，后者决定模型能不能真的去查数据库、调接口、干活。两者都属于「把大模型从聊天框里解放出来」的关键一步。

DeepSeek 的 API 兼容 OpenAI 协议，所以下面代码你换个别的兼容服务也能跑，但参数细节请以官方文档当前版本为准。先讲清楚底层，再给能抄的代码。

一、流式输出：别让用户干等

为什么需要 stream

默认情况下，你发一个请求，要等模型把整段回答生成完才一次性返回。复杂问题可能等十几秒，用户还以为卡死了。流式输出让模型边生成边往回发片段（chunk），前端可以像打字机一样逐字显示，体感快很多，也能尽早把「它确实在干活」的信号给用户。

请求怎么写

在 create 里加 stream=True：

from openai import OpenAI

client = OpenAI(api_key="你的KEY", base_url="https://api.deepseek.com")

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "用三句话介绍 TCP 三次握手"}],
    stream=True,
)

for chunk in stream:
    # 每个 chunk 里可能有一段增量文本
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

底层其实是 SSE

如果你不用 SDK 而想自己解析，流式走的是 Server-Sent Events（SSE）：服务端一行行推 data: {...} 帧，最后一帧是 data: [DONE]。每个帧里的 choices[0].delta.content 是增量文本。理解这点，你自己封装客户端或排查「为什么收不全」时就知道该去解析流式帧，而不是等一个完整 JSON。

几个容易踩的细节

chunk 可能为空。流刚开始或结束时有控制帧，没有 content，直接 print 会报错，务必加 if delta 判断。
stream=True 时拿不到完整对象。想做字数统计、落库，得自己在循环里拼接 full_text += delta。
异常发生在流中间。普通请求报错会直接抛异常；流式请求可能前几段正常、中段断流。建议用 try/except 包住整个循环，并在断流时给用户一个「生成中断」提示，而不是页面卡死。
记得 flush。Web 框架里要把增量实时推给前端，否则缓冲会让流式失去意义。
别忘了关流。有些 SDK 需要显式 stream.close()，异常分支尤其容易漏，导致连接泄漏。

流式只是传输方式，不改答案本身。同样的问题，流式和非流式拿到的最终文本应当一致（忽略采样随机性）。它解决的是「等待体验」，不是「回答质量」。

二、函数调用：让模型「动手」

模型本身不能查天气、不能读你的数据库、不能发短信。函数调用就是给它一张「工具清单」，它判断需要时，返回一个调用指令，由你的程序真正去执行，再把结果喂回去。这一步让模型从一个「嘴强王者」变成能操作真实系统的 Agent。

请求结构

核心是 tools 字段，描述每个函数的名字、用途、参数 JSON Schema：

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，如 上海"},
            },
            "required": ["city"],
        },
    }
}]

resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "上海今天适合跑步吗？"}],
    tools=tools,
    tool_choice="auto",  # 让模型自己决定是否调用
)

注意 description 和参数的 description 都很重要——模型靠这些文字判断「该不该调、传什么」。描述含糊，模型就容易调错或瞎传参。

解析模型的调用意图

模型不会直接执行，而是返回 tool_calls：

msg = resp.choices[0].message
if msg.tool_calls:
    for call in msg.tool_calls:
        name = call.function.name          # "get_weather"
        args = json.loads(call.function.arguments)  # {"city": "上海"}
        result = do_actual_call(name, args)  # 你的真实函数

这一步是关键分水岭：do_actual_call 是你写的，模型只负责「决定调哪个、传什么参数」。所以安全边界完全在你手里——模型永远碰不到真实系统，它只产出一串「我想调这个」的指令，你核准后才执行。这对权限控制和审计至关重要。

把结果喂回去，完成多轮

拿到真实结果后，要把「模型的那条指令」和「你的返回」一起塞回 messages，再请求一次：

messages = [
    {"role": "user", "content": "上海今天适合跑步吗？"},
    msg,  # 模型要求调用工具的回复（含 tool_calls）
    {
        "role": "tool",
        "tool_call_id": call.id,
        "content": json.dumps({"temp": 22, "weather": "晴", "pm25": 30}),
    },
]
final = client.chat.completions.create(
    model="deepseek-chat", messages=messages, tools=tools
)
print(final.choices[0].message.content)

注意 tool_call_id 必须和模型返回的 call.id 对上，否则服务端会报结构错误。这是新手最常见的 400 报错来源之一。

多个工具一起给

你可以一次给好几个工具，让模型自己挑。比如同时给 get_weather 和 query_order，用户问「我的订单到哪了」它就会调订单查询，问天气就调天气，无需你硬编码分支。工具越多，模型的「决策空间」越大，但也更容易选错，所以工具描述要写得界限分明。

三、流式 + 函数调用一起用

两者可以叠加：流式输出思考过程，同时模型决定调工具。解析时逻辑更复杂，因为流里可能交替出现普通文本和 tool_calls 片段。

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools,
    stream=True,
)

for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
    if delta.tool_calls:
        # 流式下 tool_calls 可能被拆成多段，需要按 index 拼接
        for tc in delta.tool_calls:
            print(f"[模型想调工具: {tc.function.name}]")

流式下的 tool_calls.arguments 是增量片段，不能每帧直接 json.loads，要按 index 累积完整后再解析。这是最容易翻车的点，务必缓存拼接。一个常见 bug 是每收到一段就尝试解析，结果拿到半截 JSON 直接炸。

四、容错：生产环境不能省

Demo 能跑不等于生产能扛。下面这些容错，接真实业务前请补齐。

1. 函数参数未必合法

模型可能返回 JSON Schema 之外的字段、类型不对、必填缺失。执行前要做校验：

try:
    args = json.loads(call.function.arguments)
    if "city" not in args:
        raise ValueError("缺 city")
except Exception as e:
    # 把错误也喂回模型，让它纠正
    return {"error": str(e)}

2. 工具执行可能超时/失败

do_actual_call 里要有超时和降级。比如查天气接口挂了，别让整个对话卡死，返回「暂时查不到」让模型兜底回复。外部依赖永远不可靠，你的工具层必须假设它会挂。

3. 防止无限循环调用

模型可能一轮调工具、拿到结果又调、又调……务必加最大轮数限制（如最多 5 轮工具调用），到顶就强制收尾，否则可能烧光配额还跑不出结果。

4. 不要用 `tool_choice` 强行指定未知函数

tool_choice={"type":"function","function":{"name":"x"}} 会强制模型调某个函数。但如果你传了模型不认识的名字，行为未定义。不确定时用 "auto"。

5. 工具返回别太大

把整个数据库查询结果塞回 messages 会撑爆上下文、烧 token。工具返回要做裁剪，只回模型需要的字段。

补充：不依赖 SDK 自己解析流式响应

有些环境装不了 OpenAI SDK，或者你想完全掌控请求过程。这时可以用 requests 直接打流式接口，手动解析 SSE 帧：

import requests, json

resp = requests.post(
    "https://api.deepseek.com/v1/chat/completions",
    headers={"Authorization": "Bearer 你的KEY", "Content-Type": "application/json"},
    json={"model": "deepseek-chat",
          "messages": [{"role": "user", "content": "你好"}],
          "stream": True},
    stream=True,
)
for line in resp.iter_lines():
    if not line:
        continue
    text = line.decode("utf-8")
    if text.startswith("data: "):
        payload = text[6:]
        if payload.strip() == "[DONE]":
            break
        chunk = json.loads(payload)
        delta = chunk["choices"][0]["delta"].get("content")
        if delta:
            print(delta, end="", flush=True)

关键点：每行以 data: 开头，末尾 data: [DONE] 表示流结束，中间帧解析后取 delta.content。自己解析能避开 SDK 版本差异，也方便在代理、内网环境里加自定义逻辑；代价是要自己处理超时、异常和断线重连，工作量比用 SDK 大。

关于 token 用量统计： 流式响应是分片的，想在结束时拿到总 token，一种做法是循环里累加每个 delta 长度做近似；另一种是等最后一个 chunk 带 usage 字段——很多实现需要在请求时带 stream_options={"include_usage": True} 才会返回（以官方文档当前版本为准）。别假设一次流式调用默认就带总用量。

补充：函数调用的安全边界

函数调用把模型从「说话」变成「动手」，安全边界必须自己守。阿文的三条铁律：第一，工具函数只做最小权限的事，比如只读查询，绝不轻易给「执行任意 shell」「删除文件」这种万能工具；第二，所有参数先校验再执行，模型可能传奇怪值、越权字段；第三，写文件、发消息这类有副作用的操作，最好加一层人工确认或白名单。把模型当成一个热情但会犯错的实习生，权限收紧，你才不会被它坑。

并行 tool_calls 的处理： 模型一次可能返回多个 tool_calls，你的执行循环要能逐个处理，并且每个结果都要带对应的 tool_call_id 回填。漏填任何一个，服务端都会拒绝整条消息。建议用字典按 id 收集结果，再统一回填，避免顺序错乱。

工具描述写得好不好，直接决定调用准不准。 模型的 description 是它的「使用说明书」。写得模糊，比如「处理数据」，模型就不知道什么时候该调；写得太像另一个工具，它就会调错。好的描述要说明「这个工具能解决什么具体问题、参数是什么含义、什么情况下该用」。花十分钟打磨工具描述，比事后修一百次错调用划算。

典型翻车场景： 一是模型「幻觉」出一个你没定义的工具名，这时要兜底返回「未知工具」并让它换思路；二是参数类型不符，比如该传数字它传了字符串，执行前 try/except 兜住；三是循环调用停不下来，必须靠最大轮数上限掐断。把这些场景提前写进容错，生产才稳。

一个实用建议： 把工具调用的「决策权」和「执行权」分开。模型只产出调用意图，你在中转层做一层白名单和参数清洗，再真正执行。这一层薄薄的网关，能挡掉绝大多数危险和无效调用，是函数调用上生产的必经之路。

小结一下这一节： 函数调用真正危险的不是模型太聪明，而是你放权太大。权限收紧、参数校验、人工确认这三道闸，少一道都可能出事。把它当成上线 checklist，每次加新工具都过一遍，比事后救火轻松得多。

五、什么场景用哪个

你的需求	用流式	用函数调用
聊天机器人逐字显示	是	否
后台批量摘要	否（省事）	否
智能客服查订单	可选	是
Agent 自动跑流程	可选	是（核心）
内部脚本调用	否	按需

收尾前的两个检查项

在上线任何接了流式或函数调用的功能前，阿文建议你做两个检查：第一，打一条原始响应，确认 reasoning_content、tool_calls 的字段名和结构与你代码里取的一致，SDK 升级常改这些；第二，用错误输入（缺参数、坏 JSON、超时）各跑一遍，确认容错真的兜住了，而不是只在正常路径验证过。生产环境的坑，九成出在没被测过的异常分支。

关于日志： 流式片段和工具调用意图都值得落日志，出问题时你能还原「模型到底想干嘛、我执行了什么」。但日志里别存密钥和用户隐私，脱敏后再记。可观测性不是锦上添花，是线上排障的救命绳——没有它，你连「为什么这条请求挂了」都要靠猜。

把流式当「体验优化」，把函数调用当「能力扩展」。两者都建在「先读懂返回结构、做好拼接与容错」的基础上。代码贴回去之前先在本地打几条原始响应确认字段，比上线后半夜救火划算得多。工程化接模型，难的从来不是发请求，而是把「可能失败、可能乱回、可能循环」的边界都兜住。

阿文上手体验