Model IO模型调用

Model I/O介绍

Model I/O 模块是与语言模型（LLMs）进行交互的 核心组件 ，在整个框架中有着很重要的地位。
所谓的Model I/O，包括输入提示(Format)、调用模型(Predict)、输出解析(Parse)。分别对应着
Prompt Template ， Model 和 Output Parser 。

简单来说就是输入、处理、输出。

Model 调用

使用配置文件方式（.env）、langchain API（langchain统一的API调用方式）

调用

1. 对话模型（Chat model）、
2. 非对话模型（LLMs、Text Model）、
3. 嵌入模型（Embedding model）。

对话模型

ChatModels，也叫聊天模型、对话模型，底层使用LLMs。
大语言模型调用，以 ChatModel 为主！

主要特点如下：
输入：接收消息列表 List[BaseMessage] 或 PromptValue ，每条消息需指定角色（如SystemMessage、HumanMessage、AIMessage）
输出：总是返回带角色的 消息对象 （ BaseMessage 子类），通常是 AIMessage。

原生支持多轮对话。通过消息列表维护上下文（例如： [SystemMessage, HumanMessage,AIMessage, ...] ），模型可基于完整对话历史生成回复。

适用场景：对话系统（如客服机器人、长期交互的AI助手）

import os
import dotenv
from langchain_openai import OpenAI

dotenv.load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")
os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")

###########核心代码############
llm = OpenAI()
str = llm.invoke("写一首关于春天的诗") # 直接输入字符串，返回的是str
print(str)

OpenAI 是非对话模型。

非对话模型

LLMs，也叫Text Model、非对话模型，是许多语言模型应用程序的支柱。主要特点如下：

输入：接受 文本字符串 或 PromptValue 对象
输出：总是返回 文本字符串

适用场景：仅需单次文本生成任务（如摘要生成、翻译、代码生成、单次问答）或对接不支持消息结构的旧模型（如部分本地部署模型）（ 言外之意，优先推荐ChatModel ）
不支持多轮对话上下文。每次调用独立处理输入，无法自动关联历史对话（需手动拼接历史文本）。

局限性：无法处理角色分工或复杂对话逻辑。

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import os
import dotenv

dotenv.load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")
os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")

########核心代码############

chat_model = ChatOpenAI(model="gpt-4o-mini")
messages = [
SystemMessage(content="我是人工智能助手，我叫小智"),
HumanMessage(content="你好，我是小明，很高兴认识你")
]

response = chat_model.invoke(messages) # 输入消息列表
print(type(response)) # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content)

ChatOpenAI是对话模型

嵌入模型

Embedding Model：也叫文本嵌入模型，这些模型将 文本 作为输入并返回 浮点数列表 ，也就是Embedding。

import os
import dotenv
from langchain_openai import OpenAIEmbeddings

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")

#############

embeddings_model = OpenAIEmbeddings(
model="text-embedding-ada-002"
)
res1 = embeddings_model.embed_query('我是文档中的数据')
print(res1)

# 打印结果：[-0.004306625574827194, 0.003083756659179926, -0.013916781172156334, ...., ]

总结

1. 使用.env的配置文件方式，将API_KEY,BASE_URL 存入，使用dotenv导入配置环境
2. 配置模型参数
3. 执行
4. 接收返回值、

Message

非对话模型输入字符串，输出字符串
对话模型输入list（BaseMessage）
BaseMessage是顶级父类，有多个子类：SystemMessage，HumanMessage，AIMessage等；

LangChain有一些内置的消息类型：

🔥 SystemMessage ：设定AI行为规则或背景信息。比如设定AI的初始状态、行为模式或对话的总体目标。比如“作为一个代码专家”，或者“返回json格式”。通常作为输入消息序列中的第一个传递。

🔥 HumanMessage ：表示来自用户输入。比如“实现 一个快速排序方法”

🔥 AIMessage ：存储AI回复的内容。这可以是文本，也可以是调用工具的请求ChatMessage ：可以自定义角色的通用消息类型FunctionMessage/ToolMessage ：函数调用/工具消息，用于函数调用结果的消息类型
其他：
ChatMessage ：可以自定义角色的通用消息类型
FunctionMessage/ToolMessage ：函数调用/工具消息，用于函数调用结果的消息类型

#1.导入相关包
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 2.直接创建不同类型消息
systemMessage = SystemMessage(
content="你是一个AI开发工程师",
additional_kwargs={"tool": "invoke_tool()"}
)

humanMessage = HumanMessage(
content="你能开发哪些AI应用?"
)

aiMessage = AIMessage(
content="我能开发很多AI应用, 比如聊天机器人, 图像识别, 自然语言处理等"
)

# 3.打印消息列表
messages = [systemMessage,humanMessage,aiMessage]
print(messages)

# 4.通过invoke参数给大模型实现传参调用，此处模型配置省略...
response = chat_model.invoke(messages)

官方文档： https://docs.langchain.com/oss/python/langchain/messages

关于上下文记忆

一次invoke，大模型只会记忆传入的Message消息作为上下文，再次使用invoke不会含有之前问答的记忆，除非传入的message包含之前的内容。

当传入多个HumanMessage时候，大模型只会回答最后一个HumanMessage，其他当作历史问过的内容不做处理。

当传入多个SystemMessage时候，大模型会记住所有内容。

模型调用方法

为了尽可能简化自定义链的创建，我们实现了一个"Runnable"协议。许多LangChain组件实现了Runnable 协议，包括聊天模型、提示词模板、输出解析器、检索器、代理(智能体)等。Runnable 定义的公共的调用方法如下：

invoke : 处理单条输入，等待LLM完全推理完成后再返回调用结果

stream : 流式响应，逐字输出LLM的响应结果

batch : 处理批量输入

这些也有相应的异步方法，应该与 asyncio 的 await 语法一起使用以实现并发：
astream : 异步流式响应
ainvoke : 异步处理单条输入
abatch : 异步处理批量输入
astream_log : 异步流式返回中间步骤，以及最终响应
astream_events : （测试版）异步流式返回链中发生的事件（在 langchain-core 0.1.14 中引入）

流式输出与非流式输出

在Langchain中，语言模型的输出分为了两种主要的模式：流式输出与非流式输出。
下面是两个场景：

非流式输出：这是Langchain与LLM交互时的 默认行为 ，是最简单、最稳定的语言模型调用方式。当用户发出请求后，系统在后台等待模型 生成完整响应 ，然后 一次性将全部结果返回 。举例：用户提问，请编写一首诗，系统在静默数秒后 突然弹出 了完整的诗歌。（体验较单调）在大多数问答、摘要、信息抽取类任务中，非流式输出提供了结构清晰、逻辑完整的结果，适合快速集成和部署。

import os
import dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
#初始化大模型
chat_model = ChatOpenAI(model="gpt-4o-mini")
# 创建消息
messages = [HumanMessage(content="你好，请介绍一下自己")]
# 非流式调用LLM获取响应
response = chat_model.invoke(messages)
# 打印响应内容
print(response)

流式输出：一种 更具交互感 的模型输出方式，用户不再需要等待完整答案，而是能看到模型 逐个token 地实时返回内容。
举例：用户提问，请编写一首诗，当问题刚刚发送，系统就开始 一字一句 （逐个token）进行回复，感觉是一边思考一边输出。更像是“实时对话”，更为贴近人类交互的习惯，更有吸引力。适合构建强调“实时反馈”的应用，如聊天机器人、写作助手等。Langchain 中通过设置 streaming=True 并配合 回调机制 来启用流式输出。

import os
import dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
# 初始化大模型
chat_model = ChatOpenAI(model="gpt-4o-mini",
streaming=True # 启用流式输出
)
# 创建消息
messages = [HumanMessage(content="你好，请介绍一下自己")]
# 流式调用LLM获取响应
print("开始流式输出：")
for chunk in chat_model.stream(messages):
  # 逐个打印内容块
  print(chunk.content, end="", flush=True) # 刷新缓冲区 (无换行符，缓冲区未刷新，内容可能不会立即显示)

print("\n流式输出结束")

批量调用

import os
import dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
# 初始化大模型
chat_model = ChatOpenAI(model="gpt-4o-mini")

messages1 = [SystemMessage(content="你是一位乐于助人的智能小助手"),
HumanMessage(content="请帮我介绍一下什么是机器学习"), ]

messages2 = [SystemMessage(content="你是一位乐于助人的智能小助手"),
HumanMessage(content="请帮我介绍一下什么是AIGC"), ]

messages3 = [SystemMessage(content="你是一位乐于助人的智能小助手"),
HumanMessage(content="请帮我介绍一下什么是大模型技术"), ]

messages = [messages1, messages2, messages3]

# 调用batch
response = chat_model.batch(messages)
print(response)

同步调用和异步调用

同步调用：

import time

def call_model():
# 模拟同步API调用
print("开始调用模型...")
time.sleep(5) # 模拟调用等待,单位：秒
print("模型调用完成。")
def perform_other_tasks():

# 模拟执行其他任务
for i in range(5):
print(f"执行其他任务 {i + 1}")
time.sleep(1) # 单位：秒

def main():
start_time = time.time()
call_model()
perform_other_tasks()
end_time = time.time()
total_time = end_time - start_time
return f"总共耗时：{total_time}秒"

# 运行同步任务并打印完成时间
main_time = main()
print(main_time)

异步调用：
异步调用，允许程序在等待某些操作完成时继续执行其他任务，而不是阻塞等待。这在处理I/O操（如网络请求、文件读写等）时特别有用，可以显著提高程序的效率和响应性。

import asyncio
import time

async def async_call(llm):
await asyncio.sleep(5) # 模拟异步操作
print("异步调用完成")

async def perform_other_tasks():
await asyncio.sleep(5) # 模拟异步操作
print("其他任务完成")

async def run_async_tasks():
start_time = time.time()
await asyncio.gather(
async_call(None), # 示例调用，使用None模拟LLM对象
perform_other_tasks()
)
end_time = time.time()
return f"总共耗时：{end_time - start_time}秒"

# # 正确运行异步任务的方式
# if __name__ == "__main__":
# # 使用 asyncio.run() 来启动异步程序
# result = asyncio.run(run_async_tasks())
# print(result)
# 在 Jupyter 单元格中直接调用

result = await run_async_tasks()
print(result)

同步异步调用大模型对比

import asyncio
import os
import time
import dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")

# 初始化大模型
chat_model = ChatOpenAI(model="gpt-4o-mini")

# 同步调用（对比组）
def sync_test():
messages1 = [SystemMessage(content="你是一位乐于助人的智能小助手"),
HumanMessage(content="请帮我介绍一下什么是机器学习"), ]
start_time = time.time()
response = chat_model.invoke(messages1) # 同步调用
duration = time.time() - start_time
print(f"同步调用耗时：{duration:.2f}秒")
return response, duration

# 异步调用（实验组）
async def async_test():
messages1 = [SystemMessage(content="你是一位乐于助人的智能小助手"),
HumanMessage(content="请帮我介绍一下什么是机器学习"), ]
start_time = time.time()
response = await chat_model.ainvoke(messages1) # 异步调用
duration = time.time() - start_time
print(f"异步调用耗时：{duration:.2f}秒")
return response, duration

# 运行测试
if __name__ == "__main__":
# 运行同步测试
sync_response, sync_duration = sync_test()
print(f"同步响应内容: {sync_response.content[:100]}...\n")

# 运行异步测试
async_response, async_duration = asyncio.run(async_test())
print(f"异步响应内容: {async_response.content[:100]}...\n")

# 并发测试 - 修复版本
print("\n=== 并发测试 ===")
start_time = time.time()

async def run_concurrent_tests():
# 创建3个异步任务
tasks = [async_test() for _ in range(3)]
# 并发执行所有任务
return await asyncio.gather(*tasks)

# 执行并发测试
results = asyncio.run(run_concurrent_tests())
total_time = time.time() - start_time

print(f"\n3个并发异步调用总耗时: {total_time:.2f}秒")
print(f"平均每个调用耗时: {total_time / 3:.2f}秒")

响应

同步调用耗时：5.73秒

同步响应内容: 机器学习是人工智能（AI）的一个子领域，旨在通过让计算机系统从数据中学习和
改进其性能，而无需明确的编程。它的核心理念是利用算法和统计模型，分析和识别数据中的模
式，从而在新数据出现时做出预测或决策。
...

异步调用耗时：4.68秒

异步响应内容: 机器学习是一种人工智能（AI）技术，它使计算机能够通过经验自我学习和改进，
而无需明确编程。换句话说，机器学习使计算机能够从数据中提取模式和规律，并根据这些信息
进行预测或决策。
机器学习的基本过程通常...

=== 并发测试 ===
异步调用耗时：3.07秒
异步调用耗时：3.61秒
异步调用耗时：7.43秒
3个并发异步调用总耗时: 7.43秒
平均每个调用耗时: 2.48秒

Prompt Template

Prompt Template，通过模板管理大模型的输入。

简介

Prompt Template 是LangChain中的一个概念，接收用户输入，返回一个传递给LLM的信息（即提示词prompt）。
在应用开发中，固定的提示词限制了模型的灵活性和适用范围。所以，prompt template 是一个 模板化的字符串 ，你可以将 变量插入到模板 中，从而创建出不同的提示。调用时：

• 以 字典 作为输入，其中每个键代表要填充的提示模板中的变量。
• 输出一个 PromptValue 。这个 PromptValue 可以传递给 LLM 或 ChatModel，并且还可以转换为字符串或消息列表。

有几种不同类型的提示模板：

• PromptTemplate ：LLM提示模板，用于生成字符串提示。它使用 Python 的字符串来模板提示。
• ChatPromptTemplate ：聊天提示模板，用于组合各种角色的消息模板，传入聊天模型。
• XxxMessagePromptTemplate ：消息模板词模板，包括：SystemMessagePromptTemplate、
• HumanMessagePromptTemplate、AIMessagePromptTemplate、
• ChatMessagePromptTemplate等
• FewShotPromptTemplate ：样本提示词模板，通过示例来教模型如何回答
• PipelinePrompt ：管道提示词模板，用于把几个提示词组合在一起使用。
• 自定义模板 ：允许基于其它模板类来定制自己的提示词模板。
  导入：

from langchain.prompts.prompt import PromptTemplate
from langchain.prompts import ChatPromptTemplate
from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.pipeline import PipelinePromptTemplate
from langchain.prompts import (
ChatMessagePromptTemplate,
SystemMessagePromptTemplate,
AIMessagePromptTemplate,
HumanMessagePromptTemplate,
)

如何插入模板：

1. 使用.format
2. 字典

# 使用位置参数
info = "Name: {0}, Age: {1}".format("Jerry", 25)
print(info)

# 使用字典解包
person = {"name": "David", "age": 40}
info = "Name: {name}, Age: {age}".format(**person)
print(info)

PromptTemplate具体使用

PromptTemplate类，用于快速构建 包含变量 的提示词模板，并通过 传入不同的参数值 生成自定义的提
示词。
主要参数介绍：
template：定义提示词模板的字符串，其中包含 文本 和 变量占位符（如{name}） ；
input_variables： 列表，指定了模板中使用的变量名称，在调用模板时被替换；
partial_variables：字典，用于定义模板中一些固定的变量名。这些值不需要再每次调用时被替
换。
函数介绍：
format()：给input_variables变量赋值，并返回提示词。利用format() 进行格式化时就一定要赋
值，否则会报错。当在template中未设置input_variables，则会自动忽略。

实例化方式1

单变量：

from langchain.prompts import PromptTemplate
# 定义模板：描述主题的应用
# 参数template，input_variables必须有
template = PromptTemplate(
template="请简要描述{topic}的应用。",
input_variables=["topic"]
)

print(template)
# 使用模板生成提示词
prompt_1 = template.format(topic="机器学习")
prompt_2 = template.format(topic="自然语言处理")
print("提示词1:", prompt_1)
print("提示词2:", prompt_2)

多变量：

#定义多变量模板
template = PromptTemplate(
template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。",
input_variables=["product", "aspect1", "aspect2"])
#使用模板生成提示词
prompt_1 = template.format(product="智能手机", aspect1="电池续航", aspect2="拍照质量")
prompt_2 = template.format(product="笔记本电脑", aspect1="处理速度", aspect2="便携性")
print("提示词1:",prompt_1)
print("提示词2:",prompt_2)

实例化方式2

from_template:（推荐）

from langchain.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template(
"请给我一个关于{topic}的{type}解释。"
)
#传入模板中的变量名
prompt = prompt_template.format(type="详细", topic="量子力学")
print(prompt)

from_template可以直接传多个参，也可以不传参

partial_variables

partial_variables:
部分提示词模板使用

方式1：

from langchain.prompts import PromptTemplate
#方式1：
template2 = PromptTemplate(
template="{foo}{bar}",
input_variables=["foo","bar"],
partial_variables={"foo": "hello"}# 部分变量foo，bar需要传入
)
prompt2 = template2.format(bar="world")# bar需要传入
print(prompt2)

方式2：

from langchain.prompts import PromptTemplate
template1 = PromptTemplate(
template="{foo}{bar}",
input_variables=["foo", "bar"]
)
#方式2：
partial_template1 = template1.partial(foo="hello")
prompt1 = partial_template1.format(bar="world")
print(prompt1)

示例：

from langchain_core.prompts import PromptTemplate
# 完整模板
full_template = """你是一个{role}，请用{style}风格回答：
问题：{question}
答案："""
# 预填充角色和风格
partial_template = PromptTemplate.from_template(full_template).partial(
role="资深厨师",
style="专业但幽默"
)
# 只需提供剩余变量
print(partial_template.format(question="如何煎牛排？"))

invoke()与format()

invoke():
只要对象是RunnableSerializable接口类型，都可以使用invoke()，替换前面使用format()的调用方
式。

format()返回值为字符串类型；
invoke()，返回值为PromptValue类型，接着调用to_string()返回字符串。

#1.导入相关的包
from langchain_core.prompts import PromptTemplate
# 2.定义提示词模版对象
prompt_template = PromptTemplate.from_template(
"Tell me a {adjective} joke about {content}."
)
# 3.默认使用f-string进行格式化（返回格式好的字符串）
prompt_template.invoke({"adjective":"funny", "content":"chickens"})

结合大模型使用

非对话大模型

import os
import dotenv
from langchain_openai import OpenAI
dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
llm = OpenAI()

对话大模型：

import os
import dotenv
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY1")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
llm = ChatOpenAI(
model="gpt-4o-mini"
)

调用过程：

prompt_template = PromptTemplate.from_template(
template = "请评价{product}的优缺点，包括{aspect1}和{aspect2}。"
)
prompt = prompt_template.format(product="电脑",aspect1="性能",aspect2="电池")
# prompt = prompt_template.invoke({"product":"电脑","aspect1":"性能","aspect2":"电池"})
print(type(prompt)) # 
# llm.invoke(prompt) #使用非对话模型调用
llm1.invoke(prompt) #使用对话模型调用

ChatPromptTemplate具体使用