模型的调用


加了a是异步

举例

给模型添加记忆功能

注意：

• append() 一次只能追加1 个元素
• "response1.content"（带双引号）：字符串字面量，文本就叫 response1.content，不会读取变量的值，不可取
• response1.content（不带引号）：访问变量的属性，拿到模型真实返回的内容

消息对象列表（字典列表vs对象列表）

注意：

消息列表必须交替出现 Human ↔ AI，不能两个 Human 挨在一起，也不能两个 AI 挨在一起： ✅ 合法顺序： SystemMessage → HumanMessage → AIMessage → HumanMessage → AIMessage

---

LangChain 两种对话格式对比

格式A：字典列表（原生OpenAI格式） 格式B：LangChain消息对象列表（SystemMessage/HumanMessage/AIMessage）

字典列表（原生dict）优点

1. 最贴近OpenRouter、OpenAI官方API，网上大部分API教程都是这个格式，没有学习额外类的成本
2. 打印出来干净，print之后直接看到role和文本，调试简单
3. 存数据库、写入文件时，可以直接json序列化，不需要额外转换

字典列表缺点

1. LangChain高级组件不识别字典：后面用到ConversationBufferMemory、LangGraph、Agent工具调用时，必须手动把字典转成LangChain消息对象，写额外转换代码
2. 没有类型校验，role写错别字（比如写成userr），Python不会报错，只会等调用API才返回失败，坑藏得深
3. 做工具调用、多模态图文消息，手动写嵌套字典很繁琐

LangChain消息对象（System/Human/AIMessage）优点

1. LangChain生态原生适配，后续写链式调用、智能Agent、LangGraph对话流程，只能用这套消息对象，不用来回做格式转换
2. 自带类型校验：比如HumanMessage不允许写错role，参数不对直接抛异常，提前发现bug（你之前遇到的ValidationError就是它的类型校验）
3. 原生支持工具调用、图片多模态消息，做复杂大模型应用更顺手
4. LangChain内置记忆组件（Memory）输出的内容，本身就是消息对象列表，可以直接塞进invoke

LangChain消息对象缺点

1. 新手要额外import、记忆3个消息类，入门门槛高一点
2. 消息对象不能直接用json.dumps保存对话记录，需要调用.model_dump()转字典才能序列化
3. print打印消息对象时，会附带一堆langchain内部属性，调试查看内容不如字典直观

给你的选型建议

✅ 如果你只是简单调用OpenRouter，做基础对话、测试、简单记忆：优先字典列表，简单好调试，不容易踩坑。 ✅ 如果你后面打算学习LangChain的Chain、Agent、LangGraph、工具调用：直接用消息对象列表，后面不用重构代码。

互相转换代码（备用，非常实用）

from langchain_core.messages import SystemMessage,HumanMessage,AIMessage
<span># 对象列表 转 字典列表</span>
<span>msg_objs</span> = [SystemMessage(content=<span>"你是音乐家"</span>),HumanMessage(content=<span>"你好"</span>)]
<span>dict_list</span> = [msg.model_dump() for msg in msg_objs]
print(dict_list)

<span># 字典列表 转 LangChain消息对象</span>
from langchain_core.messages import messages_from_dict
<span>dict_list</span> = [{<span>"role"</span>:<span>"system"</span>,<span>"content"</span>:<span>"你是音乐家"</span>},{<span>"role"</span>:<span>"user"</span>,<span>"content"</span>:<span>"你好"</span>}]
<span>msg_objs</span> = messages_from_dict(dict_list)
print(msg_objs)

---

美化输出

注意点： **为什么要写 [HumanMessage(...)]，加方括号[]

这是 LangChain 的规范： model.invoke() 这个方法，接收的参数类型是「消息列表（list）」，不是单个消息对象。

<span># ✅ 正确：[] 代表列表，里面放消息对象</span>
model.invoke(<span>[HumanMessage(content="分享一首开心的歌")]</span>)

<span># ❌ 错误：直接传单个对象，没有列表</span>
model.invoke(HumanMessage(<span>content</span>=<span>"分享一首开心的歌"</span>))

AIMessage对象里面的信息

1.content：是模型生成的最终文本答案

2.additional_kwargs：大模型返回时，除了核心回答文本 content 之外，附带返回的所有额外信息，全部塞在这个字典里。 2.1 refusal：模型是否拒绝回答，None表示同意回答

3.response_metadata：响应元数据

![[Pasted image 20260912143112.png]]

找相关信息去AIMessage去找 ![[Pasted image 20260912143645.png]]

4.id：langchain内部标识

5.tool_calls:根据调用信息

6.usage_metadata:统一消耗元数据，token消耗信息

![[Pasted image 20260912142943.png]]

---

2.stream（），流式输出

![[Pasted image 20260912143922.png]]

![[Pasted image 20260912150245.png]] 注意： 1.在print里面要加end=“”（意思就是打印结束不换行）不然就会自动换行，默认end=“\n”

2.为什么 stream 要用 for 循环？ 一句话：model.stream() 返回的是迭代器，迭代器就要用 for 循环逐个拿数据。

![[Pasted image 20260912150433.png]]

---

补充：什么是迭代器（Iterator）

迭代器：一个可以逐个产出元素的对象，遵循「惰性获取」规则——要一个，才生成一个，不会一次性把全部数据加载到内存。

在 Python 里，只要对象实现了 __iter__() 和 __next__() 方法，它就是迭代器。 model.stream() 返回的就是迭代器。

✅ 迭代器三大核心特点

1. 惰性求值（最重要！）

不会提前把所有数据准备好。 只有你调用 next()（for循环底层自动调用）的时候，它才去生成/拉取下一个元素。

对应你的 stream 场景： 不会一次性把模型全部回答拿回来。你每 next() 一次，代码就等待大模型返回下一小块 chunk。

好处：长文本不会一下子占满内存；可以做到实时流式输出。

对比列表（list）：

<span>my_list</span> = [<span>"稻"</span>, <span>"香"</span>, <span>"七"</span>, <span>"里"</span>, <span>"香"</span>]
<span># 列表创建的时候，所有元素已经全部放在内存里了</span>

迭代器：

<span># stream迭代器，一开始内存里什么文字都没有</span>
<span>stream_iter</span> = model.stream([HumanMessage(<span>"推荐歌曲"</span>)])
<span># 只有每次 next(stream_iter)，才去网络拿下一个chunk</span>

2. 只能单向遍历，一次性消费，用完就空

迭代器只能从头往后取，不能回头，不能重复遍历。一旦全部取完，迭代器就耗尽了，再取就报 StopIteration。

示例：

<span>stream_iter</span> = model.stream([HumanMessage(<span>"推荐歌曲"</span>)])
<span># 第一轮for：把迭代器里面所有chunk全部取完</span>
for chunk in stream_iter:
    print(chunk.content, <span>end</span>=<span>""</span>)

<span># 第二轮for：迭代器已经空了，什么都打印不出来！</span>
for chunk in stream_iter:
    print(chunk.content, <span>end</span>=<span>""</span>)

列表没有这个限制，列表可以反复循环很多次。

3. 不支持下标索引，不能随机访问

不能像列表一样用 stream_iter[0]、stream_iter[1] 获取第几个元素。 迭代器只能：一个接着一个按顺序拿，想拿到第3个，必须先取出第1、第2个。

---

🧪 手动演示迭代器底层（不用for循环）

for 循环本质就是自动调用 next()，捕获 StopIteration 异常。

<span># 创建迭代器</span>
my_iter = <span>iter</span>([<span>"稻"</span>, <span>"香"</span>])

<span># 手动取下一个元素</span>
<span>print</span>(<span>next</span>(my_iter)) <span># 输出 稻</span>
<span>print</span>(<span>next</span>(my_iter)) <span># 输出 香</span>

<span># 再取，没有元素了，抛出 StopIteration</span>
<span>next</span>(my_iter)

StopIteration：迭代器的「结束信号」，代表没有更多元素。 for 循环会自动捕获这个异常，然后退出循环，不用你手动写 try-except。

📊 迭代器 vs 列表（对比表，方便记忆）

📌 回到你的 LangChain stream 场景总结

model.stream() 返回迭代器：

1. 迭代器不会一次性拿到全部回答；
2. for循环不断调用next()，等待模型返回下一块 AIMessageChunk；
3. 每拿到一块，执行打印；
4. 模型生成完毕，迭代器抛出StopIteration，for循环结束。

补充一个容易混淆的概念：可迭代对象（Iterable） vs 迭代器（Iterator）

• 列表是可迭代对象，不是迭代器；可以用 iter(列表) 转成迭代器。
• 迭代器一定是可迭代对象，可迭代对象不一定是迭代器。

要不要我写一个极简的自定义迭代器类，模拟大模型stream分片输出的效果？能一眼看懂底层原理。

---

![[Pasted image 20260912152638.png]]

一句话：flush=True 强制清空输出缓冲区，立刻把文字打印到屏幕，不要等攒一批再输出。

先搞懂缓冲区是什么

Python 的标准输出有一块内存，叫输出缓冲区。 默认策略：

• 打印短内容的时候，不会直接显示到屏幕，先放到缓冲区里攒着
• 等缓冲区装满 / 遇到换行符\n / 程序结束，才一次性全部刷到屏幕上

这是操作系统为了减少 IO 次数、提升性能做的缓存机制。

放到你的流式代码场景

<span>print</span>(chunk.text, end=<span>""</span>, flush=True)

• end=""：不自动加换行，文字连续输出（解决一字一行）
• flush=True：强制刷新缓冲区，拿到一个 chunk 就立刻显示在终端

如果不加 flush=True： 因为我们写了end=""，没有换行符触发自动刷新，文字会暂存在缓冲区里，可能等好多个字攒在一起才一次性蹦出来，打字机效果会卡顿、延迟，不是实时的。

对比两种情况

1. print(chunk.text, end="") 不写 flush：文字存在缓冲区，攒一堆再显示，看起来卡顿。
2. print(chunk.text, end="", flush=True) flush=True：每 print 一次，立刻刷到屏幕，实现顺滑实时打字效果。

---

![[Pasted image 20260912153103.png]]

3.batch（）批处理

![[Pasted image 20260912153136.png]]

batch() 一次性接收多条提问，返回一个【结果列表】；列表里面装着多条回答，所以要用 for 循环逐个拿。 invoke() 只发 1 条提问，返回单个 AIMessage 对象，就一个结果，直接用，不用循环。

![[Pasted image 20260912153904.png]]

batch同时跑 3 个任务，返回3 个 AIMessage 放到一个列表里。 列表里面有多个元素，想取出每一个，就需要 for 循环遍历这个列表。

---

总结：