如果你正在用 OpenAI 的 Assistants API 做智能客服、知识库问答或者 agent 应用，大概率会在client.beta.threads.messages.create这个方法上花掉不少时间。这不是我夸张——这个接口的坑非常隐性：它本身能正常调用，返回的消息对象结构却比想象中复杂得多；而且想把它用得顺手，还绕不开 Python 里的一星*和两星**解包。这篇文章我打算把这两件事一次性讲透，既讲清楚messages.create的参数、返回值和调用时机，也讲讲解包到底在什么场景下真的有用、怎么用。

先说个真实经历。有一次我在调试一个文档问答助手，代码流程完全照着官方文档走：创建 assistant、创建 thread、调用messages.create塞入用户问题、然后打印 thread 里的消息列表等待 assistant 回复。结果列表里只有我塞进去的那条 user 消息，assistant 的回复迟迟不出现。我一度怀疑是参数传错了，反复打印返回对象，最后才发现问题根本不在messages.create本身，而是我漏掉了触发模型推理的 run 流程。这个经历让我意识到，很多人在这个接口上翻车，不是因为不会写代码，而是没搞清楚它在一个完整会话链路里到底扮演什么角色。

这篇文章适合正在用 openai Python SDK 开发助手类应用的开发者，也适合那些被返回对象层级绕晕、想搞明白*和**解包实际用法的读者。我会按照"链路位置→参数拆解→解包实战→返回解析→完整调用模式→经验总结"的顺序来写，尽量把我踩过的坑和验证过的写法都放进去。

1. 先搞清楚三件事：Thread、Message、Run，谁先谁后

1.1 messages.create 在整个链路里的真实位置

Assistants API 的三个核心概念是 Thread、Message 和 Run。我见过很多初学者把它们当成三个孤立的 API 来记忆，其实它们是一条流水线上的三个环节：Thread 是装消息的容器，Message 是容器里的一条条记录，Run 是让助手真正开始思考的那个动作。

client.beta.threads.messages.create这个方法，从名字上看是"创建消息"，从执行逻辑上看也是只负责写入消息。它做的事情就是把一条 user 或 assistant 消息追加到一个指定 thread 的历史记录里。它不负责调用模型，不负责生成回复，更不会在你调用完它之后让助手自动开口说话。

官方文档里给出的最小示例往往是这样：

message = client.beta.threads.messages.create( thread_id=thread.id, role="user", content="你需要回答什么？", )

这段代码执行后，你确实会得到一个 Message 对象，线程里也确实多了一条消息。但如果你的目标是"发完消息之后助手立刻回复"，那你会发现线程里依然只有你手动塞进去的那一条。原因很简单：回复是由模型推理产生的，而模型推理的触发入口是client.beta.threads.runs.create，不是messages.create。

1.2 一张表看懂三者的分工

为了把这三者的关系说清楚，我整理了一个对照表，这也是我在内部培训时常用的一张表：

从这个表可以很直观地看出：Messages 是 Run 的"输入"之一。当你创建一个 Run，模型会读取指定 Thread 里所有已存在的 Message，把它当作上下文来推理，然后把生成的回复以一条 role=assistant 的 Message 写回同一个线程。

所以正确的调用顺序永远是：创建 Thread → 用 messages.create 写入用户消息 → 创建 Run → 轮询 Run 状态 → Run 完成后再读取线程里新增的 assistant 消息。messages.create只是往这个流程里投喂材料的动作，它本身不产生智力活动。

1.3 为什么"消息创建成功"不等于"助手会回复"

这一点值得单独拿出来说，因为太多人在这上面反复踩坑。消息创建成功只能说明写入操作没有异常，不代表后续有任何动作。你往一个会话窗口里打了一行字但没有点发送，对方当然不会回复你。

而且这里还有一个容易被忽略的逻辑：哪怕你后面创建了 Run，Run 的执行结果也和你在创建 Run 之前到底塞了多少条消息有关系。Run 创建的那一刻，它会以当前线程里的全部历史消息作为上下文。如果你在 Run 执行期间继续用 messages.create 塞新消息，这些新消息在大多数情况下不会被当前这轮 Run 读取，而是会留到下一轮 Run 作为上下文的一部分。

我建议把"写入消息"和"触发推理"看成两个单独的动作，在业务代码里分两步来实现。曾经为了省事，我想过在创建 Run 之后立刻塞一条补充消息，期望它被当前 Run 感知到，结果实测并不是这样。后来我才意识到，Assistants API 的消息快照机制决定了 Run 使用的是触发瞬间的上下文，想要追加信息，必须等这轮 Run 结束之后再创建新一轮。

2. messages.create 的签名逐项拆解，别再把参数传错

2.1 三个必传参数：thread_id、role、content

client.beta.threads.messages.create的完整签名大致是这样的（不同 SDK 版本略有差异，但核心参数一致）：

client.beta.threads.messages.create( thread_id: str, role: Literal["user", "assistant"], content: Union[str, List[MessageContentPartParam]], attachments: Optional[List[Attachment]] = None, metadata: Optional[Dict[str, str]] = None, )

三个必传参数里，thread_id没什么好说的，就是目标会话的 ID。重点说一下role和content。

role只有两个合法取值：user和assistant。这里有个常见困惑：为什么不能传system？因为 Assistants API 的 system prompt 是在创建 assistant 时通过instructions字段设置的，不是在消息层面传入的。你在创建 Message 时传role="system"，SDK 会直接抛参数校验错误。

content是另一个容易踩坑的地方。最简单的写法是直接传字符串：

client.beta.threads.messages.create( thread_id=thread.id, role="user", content="请解释一下什么是解包？", )

这种写法在绝大多数情况下够用。但需要注意，SDK 内部会把这个字符串包装成一个带类型的消息部件数组。所以你也可以显式地传一个结构化的数组：

content=[ { "type": "text", "text": "请解释一下什么是解包？", } ]

这两种写法最终产生的效果基本一致。理解这一点很重要，因为当你想要塞入图片等多模态内容时，必须使用第二种写法，那时候你再回头看 "content 为什么设计成数组" 就很好理解了——它从一开始就是为了容纳多种消息部件而设计的。

2.2 可选参数里最容易被忽略的两个：attachments 与 metadata

attachments参数用得不多，但一旦用上就很关键。它允许你在创建消息的同时给这条消息挂上文件附件。比如让助手阅读一份 PDF，你可以这样写：

client.beta.threads.messages.create( thread_id=thread.id, role="user", content="请总结一下这份文档。", attachments=[ { "file_id": "file-XXXX", "tools": [{"type": "file_search"}], } ], )

file_id是此前通过 Files API 上传得到的文件标识，tools则指定这个附件可以被哪些工具处理。需要留意的是，并不是所有模型都支持附件，而且attachments里的工具类型需要和 assistant 配置的工具集匹配，否则可能运行时报错。

metadata是另一个非常实用的参数，它允许你在消息上挂载最多 16 对自定义键值，每对键值加起来不能超过 512 字符。这个参数对业务追踪特别有价值。比如我在做一个客服系统时，会在每条用户消息的 metadata 里写入{"session_id": "xxx", "user_tier": "vip"}，这样后续做数据分析或者消息回溯时，不需要额外维护一张映射表，直接读 metadata 就能知道消息归属。

2.3 官方示例之外：多模态 content 部件的真实写法

刚才提到 content 可以是一个消息部件数组，除了 text 类型之外，最常见的还有 image_file 类型。官方文档里对这两种类型的定义比较完整，我这里给出一个实际用过的写法：

client.beta.threads.messages.create( thread_id=thread.id, role="user", content=[ { "type": "image_file", "image_file": {"file_id": "file-XXXX"}, }, { "type": "text", "text": "这张图片里包含什么信息？", }, ], )

这里有个细节值得注意：当你传入图片文件时，最好把相关的文字描述也作为一条 text 部件放在同一个 content 数组里，模型的理解效果会好很多。我在测试中遇到过只传图片不传文字的情况，模型虽然能感知到图片存在，但缺乏明确的指令，回答经常偏题。加上一句文字指令之后，准确率明显提升。

3. 从 API 场景重新理解 * 和 **：解包不是炫技

3.1 调用时的解包：一个 * 拆序列，两个 ** 拆字典

现在来说说标题里提到的"一星 * 和两星 ** 解包"。很多人学 Python 时都见过这两个符号，但一直觉得它们只是语法糖，跟实际业务没啥关系。直到你开始大量调用 openai SDK 这类参数多、返回结构深的 API，才会真正发现解包的价值。

在函数调用场景里，*的作用是把一个可迭代对象拆成多个位置参数。打个比方，你有一个列表args = [thread.id, "user", "你好"]，执行f(*args)就等价于f(thread.id, "user", "你好")。

**的作用则是把一个字典拆成多个关键字参数。比如：

params = { "thread_id": thread.id, "role": "user", "content": "你好", } client.beta.threads.messages.create(**params)

这段代码和直接create(thread_id=..., role=..., content=...)完全等价。你可能会想：多此一举？直接写不就好了。但看下面的场景。

3.2 定义时的收集：*args 和 **kwargs 为什么叫"可变参数"

解包符号还有另一个方向的用法：定义函数时收集参数。*args会把所有多余的位置参数收集成一个元组，**kwargs会把所有多余的关键字参数收集成一个字典。这两个符号合起来，就是 Python 支持"可变参数"的底层机制。

def log_event(event_type, *args, **kwargs): print("事件类型:", event_type) print("位置参数:", args) print("关键字参数:", kwargs) log_event("message_created", thread.id, role="user")()

这个机制在封装 SDK 调用时非常有用。你可以写一个统一的助手函数，把不确定的参数全部交给**kwargs透传，这样底层 openai SDK 升级加了新参数，你的封装函数也不用改。

3.3 ** 在 messages.create 中的实战价值：条件式传参

**在messages.create场景里最大的价值是"条件式传参"。真实业务里，你往往不是每次调用都传完全相同的参数。比如同一个接口，有时要带附件，有时不带；有时要挂 metadata，有时不挂。最笨的写法是每个分支都写一遍完整调用，代码冗余且容易漏改参数。用**解包可以这么写：

base_params = { "thread_id": thread.id, "role": "user", "content": user_question, } if file_id: base_params["attachments"] = [ {"file_id": file_id, "tools": [{"type": "file_search"}]} ] if session_id: base_params["metadata"] = {"session_id": session_id} client.beta.threads.messages.create(**base_params)

这样做的好处非常明显：公共参数集中管理，可选参数按条件追加，最终调用统一展开。我曾经在一个项目里用这种方式管理十几个可能出现的参数组合，代码行数比原来的 if-else 分支版少了将近一半，而且逻辑清楚得多。

3.4 * 在处理响应列表时的优雅用法

*的实战场景更多体现在"打印"和"参数展开"上。比如你想把线程里的所有消息文本一次性打印出来，最直观的写法是：

messages = client.beta.threads.messages.list(thread_id=thread.id) texts = [m.content[0].text.value for m in messages.data if m.content] print(*texts, sep="\n---\n")

这里的*texts就是把列表里的多个文本作为多个参数传给了print。如果没有这个星号，你打印出来的会是一整串带方括号和引号的列表字面量。用了星号之后，每一条消息会以换行加分隔线的形式打印，日志阅读体验完全不一样。

另外，如果某个函数需要接收多个位置参数，而你的数据正好在一个列表里，*也是天然的解包工具：

def show_message(thread_id, role, content): print(f"[{role}] {content}") msg_data = [thread.id, "user", "你好"] show_message(*msg_data)

这行代码里的*msg_data直接把三个元素对应到了三个位置参数上。在写测试桩、批量数据回放这类场景里非常顺手。

4. 拿到返回的 Message 对象，如何正确解析嵌套内容

4.1 打印原始结构，先看清 content 的层次

messages.create返回的是一个openai.types.beta.threads.Message对象。很多人在这一步开始懵：直接print(message)会看到一堆content=[MessageContentText(...)]之类的输出，根本不知道该取哪个字段。

我的建议是遇到不确定的对象结构时，先把它的底层 JSON 打出来，一秒钟就能看清层级：

print(message.model_dump_json(indent=2))

这段代码会输出类似如下的结构：

{ "id": "msg_XXXX", "object": "thread.message", "created_at": 1735000000, "thread_id": "thread_XXXX", "role": "user", "content": [ { "type": "text", "text": { "value": "你好", "annotations": [] } } ], "file_ids": [], "assistant_id": null, "run_id": null, "metadata": {} }

看到这个结构你就明白了：content是一个列表，列表里每个元素都带一个type字段，类型为 text 的元素内部还有一个嵌套的text对象，真正的文本内容在text.value里。

4.2 提取正文文本的稳健写法

提取正文文本的写法必须考虑到 content 列表可能包含多个元素、每个元素类型可能不同的情况。我实际项目中用的解析函数是这样的：

def extract_message_text(message): if not message or not message.content: return "" parts = [] for block in message.content: if block.type == "text" and block.text: parts.append(block.text.value or "") return "\n".join(parts)

这个函数有两个关键点。第一，先判断message.content是否存在，因为某些消息对象可能出现空 content，直接索引会抛异常。第二，每次循环都要检查block.type，因为 content 里可能有 image_file 或其他类型，它们没有.text.value属性，不加判断直接访问会报 AttributeError。

4.3 处理空 content 和 None 字段的防御逻辑

在真实 API 返回中，空 content 和 None 字段比你想象中更常见。具体表现在这几个地方：

我在一次数据迁移中就吃过亏。当时批量处理历史消息，没有做空 content 判断，结果读到几条内容为空的 assistant 消息直接崩了。后来所有读取入口都统一走extract_message_text这个防御式函数，再也没出过类似问题。

4.4 attachments 与 file_ids 的读取思路

消息对象里除了 content，还有attachments和file_ids字段。前者最常见的使用场景是检查这条消息带了哪些文件，后者是旧版本 SDK 遗留的文件标识列表，现在大部分场景已经被 attachments 取代。

如果你需要解析附件信息，我建议也用同样的"先打印再取值"的思路处理。附件的 key 实际上对应的是此前上传到 OpenAI 的文件 ID，它在消息层面不包含文件内容，只包含文件引用。想要获取文件名、大小这些信息，需要额外调用文件检索接口。这一点容易误导人——你拿到一个 file_id 以为能直接读内容，其实它只是一个引用凭证。

5. 消息创建只是开始：一个完整的多轮对话调用模式

5.1 创建消息之后，必须补上 run 才能拿到回复

前面已经说过，messages.create之后要拿到 assistant 回复，必须创建 Run。一个完整的流程是这样：

1. 创建 Thread
2. 用messages.create写入用户消息
3. 用client.beta.threads.runs.create发起推理
4. 轮询runs.retrieve直到 Run 状态变为 completed
5. 用messages.list读取新增的 assistant 消息

这个过程我第一次跑通时犯了一个低级错误：创建 Run 之后直接sleep(2)然后读取消息，结果经常扑空。后来才发现 Run 的完整状态流转是queued → in_progress → completed，模型推理可能要十几秒甚至更久。轮询才是正确处理方式，固定 sleep 不靠谱。

5.2 最小可运行的对话循环代码

给你一个我当时沉淀下来的最小可运行调用模式，经过多次验证，可以直接抄走用：

import time def ask_assistant(assistant_id, thread_id, user_question): # 第 1 步：写入用户消息 client.beta.threads.messages.create( thread_id=thread_id, role="user", content=user_question, ) # 第 2 步：创建 Run run = client.beta.threads.runs.create( thread_id=thread_id, assistant_id=assistant_id, ) # 第 3 步：轮询 Run 状态 while run.status in ("queued", "in_progress", "requires_action"): run = client.beta.threads.runs.retrieve( thread_id=thread_id, run_id=run.id, ) time.sleep(1) if run.status != "completed": raise RuntimeError(f"Run 未成功完成，状态: {run.status}") # 第 4 步：读取消息，取最后一条 assistant 消息 messages = client.beta.threads.messages.list( thread_id=thread_id, order="desc", ) for m in messages.data: if m.role == "assistant": return extract_message_text(m) return ""

这里有两个细节值得说一下。第一，order="desc"按时间倒序返回，最新消息排在最前面，所以循环里命中的第一条 assistant 消息就是最新回复。第二，requires_action状态也要算在轮询范围内，因为如果助手配置了函数调用，它会在等待工具执行结果时停留在这个状态，直接跳过的话会导致工具调用链路断裂。

5.3 用字典 + ** 统一管理多轮会话参数

多轮对话场景下，每轮用户消息的附加参数可能都不一样。比如第一轮不需要附件，第二轮用户上传了一张图片。用传统写法，你得写两个很长的调用分支。用字典加**解包，代码可以收敛成下面这样：

def add_dialog_message(thread_id, role, content, attachment=None, meta=None): call_params = { "thread_id": thread_id, "role": role, "content": content, } if attachment: call_params["attachments"] = [attachment] if meta: call_params["metadata"] = meta return client.beta.threads.messages.create(**call_params)

调用侧只需要按需传入非默认参数：

add_dialog_message(thread.id, "user", "请分析这张表", attachment=None) add_dialog_message( thread.id, "user", "这是新上传的图片，请描述", attachment={"file_id": file_id, "tools": [{"type": "file_search"}]} )

这种模式在参数组合多变的时候特别稳，你永远不会忘了某个分支要传什么参数，因为公共参数都在字典里统一维护。

5.4 错误处理与超时重试的实用策略

messages.create本身是一个轻量接口，但放在整个对话链路里，它依然有可能因为网络抖动、参数校验失败而报错。我的实用策略有三条。

第一，创建消息这步可以做简单重试。因为消息创建是幂等的，重复创建虽然会产生多条一样的历史消息，但在重试场景下影响不大。不过要注意，重试机制只应该包裹"创建消息"这步，不要包裹"创建 Run"。Run 的执行不是幂等的，重复创建 Run 会导致同一问题被模型同时处理多次，白白消耗 token。

第二，在轮询 Run 时设置超时上限。我一般在测试时用 30 秒，生产环境放到 60 秒，超过时间仍然没有 completed，就进入降级逻辑，返回"服务繁忙"之类的响应。

第三，对返回的 Message 对象做空值防御。因为创建消息虽然成功，但极少数情况下返回对象里 content 仍然可能为空。解析函数里先判空，再取值，能避免线上告警被一堆 AttributeError 刷屏。

6. 真实项目里积累下来的几条经验

6.1 用 metadata 给消息打上业务标签

我在前面提过 metadata 的用法，这里再展开说一句。给消息打业务标签是我在真实项目里验证过最有价值的习惯之一。比如你是做客服系统的，给每条用户消息在 metadata 里写上conversation_id、user_id，后续如果需要做离线数据分析或者手动排查问题，直接按 metadata 过滤即可，不需要在业务侧维护额外的关联表。

还有个更实用的场景：当你把一段长对话归档或迁移时，metadata 里的标签能帮你快速区分消息来自哪个渠道、哪个版本。我见过太多团队把消息源信息放在日志里，归档之后想找一条特定消息，翻半天日志头都大了。metadata 直接在数据上挂标签，省心得多。

6.2 别把 assistant 的历史消息当成用户消息重放

多轮对话里有一个隐蔽的错误操作：为了给模型提供更多上下文，有人会把上一轮 assistant 的回复原封不动再以 user 身份塞回去。这种做法的后果是上下文里出现矛盾的角色信息，模型可能把"自己的话"误当成"用户的指令"，回答质量直线下降。

正确的做法是把整个线程历史留在 Thread 里。Assistants API 的优势就在于它会自动把所有历史消息作为上下文，你不需要手动搬运消息。如果你确实需要注入额外语境，优先选择更新 assistant 的instructions，而不是伪造用户消息。我认为这是很多人在消息创建环节犯过的最隐蔽错误之一，值得重视。

6.3 beta 接口的兼容性意识：锁版本、看 changelog

client.beta.threads.messages.create这个接口路径里写了beta，意味着它随时可能调整。OpenAI 官方已经在主推更新的 Responses API，Assistants API 虽然仍可用，但新功能迭代速度确实在放缓。

我的建议是生产项目里锁死 openai SDK 版本，不要用浮动的大版本号直接拉最新。SDK 升级前先仔细看 changelog，确认接口签名没有破坏性变更。我曾经在一次大版本升级后遇到过messages.create的 attachments 参数结构调整，导致线上创建消息时工具调用不生效。排查了半天，最后发现是 SDK 版本不一致导致的行为差异。从那以后，SDK 版本变更必须走测试流程。

6.4 最后分享一个小技巧：用解包写测试桩

文章最后分享一个让我在做接口测试时轻松很多的技巧。当你需要 mockclient.beta.threads.messages.create时，可以用**kwargs写一个万能的假实现：

def fake_create(**kwargs): print("调用参数:", kwargs) return { "id": "msg_fake", "role": kwargs.get("role"), "content": [{"type": "text", "text": {"value": kwargs.get("content"), "annotations": []}}], }

这样无论测试里传哪些参数，fake 函数都能接收并打印，你据此可以快速确认调用侧到底传了什么参数、有没有遗漏必填项。配合*args还能处理一些依赖位置参数的变体调用。这个写法在写单元测试时价值极高，自从用了这种 mock 模式之后，我再也没有在"参数到底传没传对"这个问题上浪费过排查时间。

说到底，messages.create本身的用法并不复杂，复杂的是它嵌套的返回结构和它与 Run 之间的配合关系。而*和**解包，本质上就是帮助你在面对这种嵌套结构时，用更少的代码写出更灵活的调用逻辑。把这两个点都吃透之后，你的助手应用开发效率应该会有一个明显的提升。