Python调用文心一言API实战：鉴权、多轮对话与流式输出详解

大模型API接入这件事，说穿了就是三件事：拿到钥匙、把话递过去、把回的话接住。但真到自己动手写第一行代码的时候，很多人会卡在几个很具体的地方——鉴权参数怎么拼、返回结构长什么样、多轮对话的上下文到底谁来维护、流式输出为什么接了半天没反应。这篇就围绕“用Python调用文心一言”这个具体场景，把我自己从零跑通到实际用起来这一路的东西完整摊开讲一遍，包括那些文档里写得比较简略、但实际写代码时一定会撞上的细节。

适合谁看：有Python基础、想在自己的脚本或小工具里接一个中文对话能力的开发者；也适合已经跑通过一次Demo、但想搞清楚“为什么这么写”“怎么写得稳一点”的人。下面所有代码都是可以直接复制运行的，参数和结构我会逐个解释清楚，不会只丢一段代码让你自己猜。

1. 动手之前先把这几件事想明白

1.1 为什么选API调用而不是别的路子

在Python里用上文心一言，路径其实不止一条：可以用官方提供的Python SDK，也可以直接发HTTP请求。我一开始图省事用的是SDK，后来在几个项目里都改成了直接发HTTP请求，原因很实际——SDK本质上是对HTTP接口的一层封装，多一层封装就多一层版本依赖，一旦SDK升级或者和你项目里的其他依赖打架，排查起来很烦。直接发请求的话，你只需要一个能发POST的库，请求体长什么样、返回什么结构，全都在你眼皮底下，出问题一眼就能看到是哪一层的事。

当然这不是说SDK不能用。如果你只是快速验证一个想法，SDK确实能省掉拼参数的时间。但只要你打算把这个能力放进一个要长期维护的项目里，我建议还是走HTTP直连，把请求和解析这两段逻辑自己攥在手里。这篇后面的代码也全部基于直连的方式来讲，用到的库就是最普通的requests，没有额外依赖。

还有一个前置认知要建立起来：这类对话接口本质上是无状态的。也就是说，服务端不记得你上一句说了什么，每一次请求都是独立的。所谓“多轮对话”，是你自己在客户端把历史消息攒起来，每次请求时一起发过去。这个点如果一开始没想清楚，后面写多轮逻辑的时候会绕很多弯路。

1.2 鉴权这件事：access_token是怎么来的

调用接口需要一个凭证，通常叫access_token。这个东西不是你在控制台复制一段固定字符串就完事的，它有有效期，需要用一个长期凭证去换。整个链路是这样的：你在平台上拿到一对API Key和Secret Key，用这两个去请求一个专门的鉴权地址，换回一个有时效的access_token，然后每次调对话接口时把这个token带上。

这里有个非常容易踩的坑：token是有有效期的，一般是30天。很多人写Demo的时候把token硬编码在代码里，跑通了就以为万事大吉，结果过了一个月发现接口全报鉴权失败，回头查半天才发现是token过期了。正确的做法是在代码里实现“获取+缓存+过期刷新”这一套逻辑，让程序自己管理token的生命周期。

换取token的请求长这样，用requests发一个POST：

import requests API_KEY = "你的API Key" SECRET_KEY = "你的Secret Key" def get_access_token(api_key, secret_key): url = "https://aip.baidubce.com/oauth/2.0/token" params = { "grant_type": "client_credentials", "client_id": api_key, "client_secret": secret_key } resp = requests.post(url, params=params, timeout=10) data = resp.json() if "access_token" not in data: raise RuntimeError(f"获取token失败: {data}") return data["access_token"], data.get("expires_in", 2592000)

注意grant_type这个参数固定是client_credentials，这是OAuth体系里的标准写法，意思是“我用客户端凭证换访问令牌”。expires_in返回的是有效期秒数，通常接近30天。拿到之后建议连同获取时间一起存起来，下次用之前先判断有没有快过期，快过期就重新换一个，而不是每次都去请求——频繁换token既没必要，也可能触发平台的频率限制。

提示：API Key和Secret Key属于敏感信息，绝对不要写死在会提交到代码仓库的文件里。用环境变量或者本地配置文件读取，配置文件记得加进忽略列表。

1.3 请求体的基本骨架

对话接口的请求体是一个JSON，核心字段其实不多，但每个都有讲究。我先把一个最小可用的结构摆出来，然后逐个说：

{ "messages": [ {"role": "user", "content": "你好，介绍一下你自己"} ] }

messages是一个数组，里面每个元素是一条消息，role表示说话的角色，content是内容。角色一般有三种：user是你（用户）说的话，assistant是模型回的话，system是用来设定模型行为的一段前置指令。多轮对话就是把历史里user和assistant的消息按顺序都塞进这个数组，最后一条是当前这轮user的新问题。

这里有个细节值得说：system角色的消息不是必须的，但用好了能让输出稳定很多。比如你希望模型始终用简洁的、不带客套话的风格回答，就可以在messages最前面放一条system消息把规矩定下来。我实测下来，有明确system指令和没有，输出的风格差异是肉眼可见的。

2. 把第一次调用真正跑通

2.1 一个能直接运行的最小示例

把前面的token逻辑和请求体拼起来，就是一个完整的最小调用。这里我用的是文心一言的对话补全接口，模型名称按你实际开通的填，下面用ernie-bot作为示例：

import requests def chat_once(access_token, user_input, model="ernie-bot"): url = ( "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/" f"{model}?access_token={access_token}" ) payload = { "messages": [ {"role": "user", "content": user_input} ] } headers = {"Content-Type": "application/json"} resp = requests.post(url, json=payload, headers=headers, timeout=30) return resp.json() if __name__ == "__main__": token, _ = get_access_token(API_KEY, SECRET_KEY) result = chat_once(token, "用一句话解释什么是递归") print(result)

跑通之后你会看到一个JSON返回，里面最关键的是result字段，那就是模型生成的文本。除此之外还有usage字段记录token消耗，id是这次请求的标识。第一次跑建议直接把整个返回print出来看一遍，把结构记在脑子里，后面解析的时候心里有数。

这里有个新手常犯的错：把access_token拼在URL的查询参数里，而不是放在请求头。这个接口的设计就是走查询参数的，你放请求头反而会鉴权失败。不同平台的约定不一样，接一个新接口时先看清楚它要求token放哪儿，别想当然。

2.2 返回结构里哪些字段值得盯

返回的JSON字段不少，但日常真正要用的就那么几个。我整理成一张表，方便对照：

usage这块特别值得关注。很多人只盯着result看，忽略了token消耗，结果上下文越攒越长，某天突然发现调用成本涨得离谱，或者请求直接被拒——因为超出了模型的最大上下文长度。养成每次记录total_tokens的习惯，能帮你提前发现上下文膨胀的问题。

error_code和error_msg是排错的关键。接口出错时不会抛异常给你，而是返回一个带错误码的JSON，如果你只取result字段，出错时拿到的是None，然后程序在别的地方莫名其妙崩掉，排查起来很绕。所以解析返回时第一件事应该是判断有没有error_code。

2.3 超时和重试：别让一次网络抖动毁掉整个流程

网络请求一定要设超时，这是铁律。不设超时的话，遇到服务端响应慢，你的程序会一直挂在那里等，整个流程卡死。上面代码里我写了timeout=30，这个值对对话类接口是合理的——模型生成需要时间，设太短（比如3秒）会导致正常的长回答被误判为超时。

但光有超时还不够，还得有重试。网络抖动、偶发的服务端限流，都会让单次请求失败，这时候无脑重试往往能解决大部分问题。不过重试要讲究策略：不能失败就立刻重试，那样在服务端压力大时会雪上加霜；也不能无限重试，得有次数上限。我一般用“指数退避”——第一次失败等1秒重试，第二次等2秒，第三次等4秒，最多重试3次。这样既给了服务端喘息时间，又不至于让用户等太久。

import time def chat_with_retry(access_token, user_input, max_retries=3): for attempt in range(max_retries): try: result = chat_once(access_token, user_input) if "error_code" not in result: return result # 某些错误码重试也没用，比如参数错误，直接抛 if result["error_code"] in (17, 18): # 示例：限流类错误可重试 raise RuntimeError(result.get("error_msg")) except (requests.Timeout, requests.ConnectionError): pass time.sleep(2 ** attempt) raise RuntimeError("重试多次仍失败")

这里要区分“可重试错误”和“不可重试错误”。限流、超时、连接失败这类是临时的，重试有意义；而参数格式错误、鉴权失败这类，重试一百次结果都一样，应该直接报错让人去修。把这两类分开处理，能省掉很多无谓的等待。

3. 多轮对话的上下文到底谁来管

3.1 无状态接口下的“记忆”是怎么造出来的

前面提过接口是无状态的，那么多轮对话怎么实现？答案朴素得有点反直觉：每次请求都把完整的历史消息重新发一遍。模型本身不记忆，是你在客户端维护了一个消息列表，每次把整个列表递过去，模型基于这个列表生成下一条回复。

理解这一点之后，很多困惑就解开了。比如为什么对话越长响应越慢、越贵——因为每次请求携带的上下文越来越长，输入token在涨。再比如为什么有时候模型“忘了”前面说过的话——很可能是你的历史列表在某处被截断或者没正确拼接。

一个典型的多轮维护逻辑长这样：

class ChatSession: def __init__(self, access_token, system_prompt=None): self.access_token = access_token self.messages = [] if system_prompt: self.messages.append({"role": "system", "content": system_prompt}) def send(self, user_input): self.messages.append({"role": "user", "content": user_input}) result = chat_once_with_messages(self.access_token, self.messages) reply = result["result"] self.messages.append({"role": "assistant", "content": reply}) return reply

关键在最后两步：拿到回复后，要把这条assistant消息也追加进历史列表。很多人第一次写多轮时漏了这一步，结果模型永远只看到用户说的话，看不到自己之前回过什么，对话就变成了“每轮都从头开始”。这个坑非常隐蔽，因为单轮测试完全正常，只有连续对话时才暴露。

3.2 上下文长度控制：什么时候该“忘掉”早期内容

历史列表不能无限增长。每个模型都有最大上下文长度限制，超了就会报错或者被截断。所以必须有一套“遗忘”策略。最简单的做法是只保留最近N轮对话，更早的丢掉。但直接粗暴截断有个问题：如果早期对话里有关键信息（比如用户一开始说的需求背景），丢掉之后模型就接不上了。

我的做法是分层处理：system消息永远保留在最前面不动；中间的历史按轮次从旧到新保留，超过阈值就从最旧的开始删；同时给每轮对话估算token数，控制在模型上限的70%左右留出余量。为什么留余量？因为模型生成回复也要消耗token，如果输入已经顶满上限，输出就没空间了，会直接报错。

估算token有个粗略经验：中文大约1个字对应1到2个token，英文大约4个字符对应1个token。精确计算需要专门的tokenizer，但日常控制长度用粗估就够了，宁可保守一点。

3.3 一个容易忽略的细节：消息顺序和角色交替

messages数组里的消息顺序是有意义的，必须按时间先后排列。而且角色一般是user和assistant交替出现，system只在最前面。如果你把顺序搞乱，或者连续放两条user消息中间没有assistant，模型的表现会变得很奇怪——它可能把两条user消息当成一个整体，或者对角色产生混淆。

还有一种情况：有些场景下你想“重新开始”一段对话，但保留system设定。这时候不要新建一个空列表把system也丢了，而是保留system那条，清掉后面的user和assistant。这个细节在写带“清空对话”按钮的界面时特别容易出错。

4. 流式输出：为什么接了没反应

4.1 流式和非流式的本质区别

非流式是等模型把整段话生成完，一次性返回；流式是模型边生成边往回吐，前端可以一个字一个字地显示出来。体验上流式明显更好，尤其是回答较长的时候，用户不用干等。但流式的代码结构和非流式差别不小，很多人照着非流式的思路去接流式，结果发现“没反应”或者“只拿到最后一段”。

流式的关键在于：返回的不是一个完整的JSON，而是一串按行分隔的数据块（通常叫SSE，Server-Sent Events格式）。每个数据块是一小段增量内容，你需要逐块读取、逐块解析、把增量拼起来。请求时要在payload里加一个开关字段，通常是"stream": true。

def chat_stream(access_token, messages, model="ernie-bot"): url = ( "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/" f"{model}?access_token={access_token}" ) payload = {"messages": messages, "stream": True} headers = {"Content-Type": "application/json"} with requests.post(url, json=payload, headers=headers, stream=True, timeout=60) as resp: for line in resp.iter_lines(): if not line: continue line = line.decode("utf-8") if line.startswith("data:"): chunk = line[5:].strip() if chunk == "[DONE]": break yield chunk

注意requests.post这里必须加stream=True，否则iter_lines拿不到增量，会等整个响应结束才一次性给你。这是“接了没反应”最常见的原因——请求发出去了，但客户端在等完整响应，看起来就像卡住了。

4.2 增量拼接和结束标志

每个数据块解析出来是一个JSON，里面有个字段装着这一小段的文本增量（字段名各平台略有差异，常见是result）。你要做的是把这些增量按顺序拼起来，拼完就是完整回复。同时要注意结束标志，通常是收到一个特殊的[DONE]或者某个字段变成true，收到之后就该停止读取了。

拼接的时候有个坑：增量是按“字符片段”给的，可能把一个词甚至一个多字节字符拆开。如果你在拼接过程中做编码转换不当，会出现乱码。稳妥的做法是统一用UTF-8解码，拼完再处理，不要在中间环节反复转码。

还有一个实际使用中的经验：流式输出时，如果中途网络断了，你已经收到的部分内容是有效的，但回复不完整。所以流式场景下要么做好“断点续传”的提示，要么在界面上明确告诉用户“回复可能不完整”。非流式就没这个问题，要么全有要么全无。

4.3 流式下的错误处理更麻烦

非流式出错时返回一个带错误码的JSON，好判断。流式出错时，错误可能出现在流的中间——前面正常吐了几段，突然来一个错误块。这时候你的解析逻辑要能识别出“这一块不是正常内容而是错误信息”，不能傻乎乎地把它当正文拼进去。

我的处理方式是：每个数据块解析后先检查有没有错误标识字段，有的话记录错误、中断循环、把已收到的内容标记为不完整。这样至少不会把错误信息当成模型说的话展示给用户，那种体验非常糟糕。

5. 实际用起来才会遇到的几个问题

5.1 参数调优：temperature和top_p到底怎么设

除了messages，请求体里还能带一些控制生成风格的参数，最常用的是temperature和top_p。这两个都是控制“随机性”的，但机制不同。temperature越高，输出越发散、越有创意；越低越保守、越确定。top_p是从概率分布里取一个累积阈值，只从高概率的候选里选。

实际怎么设？我的经验是按场景分：

有个原则：temperature和top_p一般不要同时调到很高，两个都高会让输出变得非常不可控。调的时候一次只动一个，观察效果，别两个一起改，否则出了问题都不知道是哪个参数导致的。

5.2 内容安全过滤导致的“空回复”

有时候请求成功了，error_code也没有，但result是空的，或者返回一句“抱歉，我无法回答这个问题”。这通常是触发了内容安全过滤。这类过滤是平台侧做的，你无法绕过，也不应该试图绕过。

遇到这种情况，正确的处理是把它当成一种正常的业务分支：给用户一个友好的提示，而不是把空字符串直接展示出去。同时可以在日志里记录下触发过滤的输入（注意脱敏），方便后续分析是不是提问方式有问题。有时候换个问法、把问题拆细一点，就能正常得到回复，这属于提示词层面的优化，而不是去对抗过滤机制。

5.3 并发调用时的限流应对

如果你的程序需要同时处理多个请求（比如批量处理一批文本），很容易撞上平台的频率限制。表现是突然一批请求返回限流错误码。应对办法有两个层面：一是客户端做并发控制，别一次性把几百个请求全发出去，用信号量或者线程池限制同时进行的请求数；二是配合前面说的指数退避重试，让被限流的请求等一等再试。

我一般会把并发数控制在个位数到十几个之间，具体看平台给的配额。盲目提高并发不会让总吞吐变高，反而因为大量请求被限流、反复重试，整体效率更低。这个道理和高速公路堵车一样，车太多谁都走不动。

5.4 日志和可观测性：出问题时靠什么定位

线上跑起来之后，出问题是必然的。定位问题的前提是你有足够的日志。我建议至少记录这几样：每次请求的时间戳、请求的消息条数和估算token数、返回的id、usage、以及有没有错误码。这些信息在排查“为什么这次回复不对”“为什么成本突然涨了”的时候是救命的。

特别提醒一点：日志里不要记录完整的用户输入和模型输出原文，尤其是涉及用户隐私的场景。记录消息条数、长度、token数这些元信息就够了，原文该脱敏的脱敏。这既是合规要求，也是良好的工程习惯。

6. 把它封装成一个能复用的模块

6.1 分层设计：鉴权、请求、会话各管各的

前面零散讲了很多，真正落到项目里，我建议按职责拆成三层：鉴权层只管token的获取和刷新，请求层只管发请求和解析返回，会话层只管维护消息历史。这样拆的好处是每一层都能单独测试、单独替换。比如哪天要换一个模型或者换一个平台，只需要动请求层，会话逻辑不用碰。

鉴权层的关键是缓存。用一个模块级的变量存token和它的过期时间，每次取token前先判断，没过期就直接返回，过期了才去换。这样既避免了频繁请求，又保证了不会用到过期token。

6.2 配置外置：别把密钥写进代码

所有敏感配置——API Key、Secret Key、模型名称、超时时间、重试次数——全部外置到配置文件或环境变量。代码里只读配置，不写死值。这样做的好处是同一份代码可以在开发、测试、生产环境用不同的配置跑，不用改代码。而且密钥不会跟着代码进仓库，安全性有保障。

配置读取这块，我习惯写一个小的配置加载函数，优先读环境变量，读不到再读本地配置文件，都没有就给默认值或者报错。这样在容器化部署时用环境变量，本地开发时用配置文件，两种方式都支持。

6.3 一个完整的调用流程串起来

把前面所有东西串起来，一次完整的调用流程是这样的：程序启动时加载配置；第一次调用前检查token，没有或过期就去换；组装messages（如果是多轮，带上历史）；发请求，带超时；拿到返回先判断错误码；正常的话取result，同时记录usage；如果是多轮，把本轮user和assistant消息追加进历史；根据上下文长度决定要不要裁剪历史。这一套走下来，就是一个能稳定跑在生产环境里的最小闭环。

我在实际项目里用这套结构跑了挺长时间，中间遇到过token过期、限流、上下文超长、流式中断各种情况，因为每一层都有对应的处理，所以没有出现过整个流程崩掉的情况。这套东西不复杂，但每个环节都得想到，漏一个就可能在某个特定时刻给你来个“惊喜”。

最后分享一个我自己的小习惯：每次接入一个新接口，我都会先写一个最简单的脚本，把请求发出去、把原始返回完整打印出来，什么都不解析。等我看清楚返回长什么样了，再开始写解析和业务逻辑。这个习惯帮我省了很多“照着文档写但就是不work”的时间——因为文档和实际返回之间，偶尔会有那么一点点出入，亲眼看过才踏实。