Model I/O 之模型调用

Model I/O 模块是与语言模型（LLMs）进行交互的核心组件 ，在整个框架中有着很重要的地位，本文介绍大模型的调用。

1、模型参数

这里以 LangChain 的API为准，使用对话模型，进行测试 。

模型调用函数使用时需初始化模型，并设置必要的参数。

必须设置的参数：

base_url ：大模型 API 服务的根地址

api_key ：用于身份验证的密钥，由大模型服务商（如 OpenAI、百度千帆）提供

model/model_name ：指定要调用的具体大模型名称（如 gpt-4-turbo 、 ERNIE-3.5-8K 等）

其它参数：

temperature ：温度，控制生成文本的“随机性”，取值范围为0～1。值越低 → 输出越确定、保守（适合事实回答）,

值越高 → 输出越多样、有创意（适合创意写作）通常，根据需要设置如下：

• 精确模式（0.5或更低）：生成的文本更加安全可靠，但可能缺乏创意和多样性。

• 平衡模式（通常是0.8）：生成的文本通常既有一定的多样性，又能保持较好的连贯性和准确性。

• 创意模式（通常是1）：生成的文本更有创意，但也更容易出现语法错误或不合逻辑的内容。

max_tokens ：限制生成文本的最大长度，防止输出过长。

Token是什么？

​ Token 是大模型处理文本的最小单位（相当于自然语言中的词或字），输出时逐个token依次生成。

​ 大语言模型(LLM)通常也是以token的数量作为其计量(或收费)的依据。1个Token≈1-1.8个汉字，1个Token≈3-4个英文字母

max_tokens设置建议：

• 客服短回复：128-256。比如：生成一句客服回复（如“订单已发货，预计明天送达”）

• 常规对话、多轮对话：512-1024

• 长内容生成：1024-4096。比如：生成一篇产品说明书（包含功能、使用方法等结构）

2、模型参数存储

模型参数存储一般有三种方式，硬编码、环境变量和 .env ，一般建议使用 .env 方式，演示如下：

根目录创建 .env 文件，添加相关apikey 和 baseurl内容进入。

apikey="sk-ece30a3360d2*********d4efc9d0"
baseurl="https://api.deepseek.com/v1"
model="deepseek-chat"

加载 .env 文件，会把该文件的环境变量存储到该进程中，使用方式如下：

import dotenv

dotenv.load_dotenv()

3、初识模型调用

模型调用示例：

from langchain_openai import ChatOpenAI
import os
import dotenv

# 加载配置文件
dotenv.load_dotenv()

os.environ["OPENAI_BASE_URL"] = os.getenv("baseurl")
os.environ["OPENAI_API_KEY"] = os.getenv("apikey")
model = os.getenv("model")

# 1、获取对话模型：
chat_model = ChatOpenAI(
    # 必须要设置的3个参数，model / api_key / base_url
    model=model  # 如果不指定，默认使用的是gpt-3.5-turbo模型
    # 当没有显式的声明base_url和api_key的时候，默认会从环境变量中查找OPENAI_BASE_URL 和 OPENAI_API_KEY
)

# 2、调用模型
response = chat_model.invoke("你好?")
print(response.content)

输出示例：

你好！很高兴见到你！😊
有什么我可以帮助你的吗？无论是回答问题、聊天交流，还是需要任何其他帮助，我都很乐意为你服务！

4、对话模型的消息

聊天模型，出了将字符串作为输入外，还可以使用聊天消息作为输入，并返回聊天消息作为输出。

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

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

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

• 🔥 AIMessage ：存储AI回复的内容。这可以是文本，也可以是调用工具的请求。

• ChatMessage ：可以自定义角色的通用消息类型

• FunctionMessage/ToolMessage ：函数调用/工具消息，用于函数调用结果的消息类型。

示例：

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

dotenv.load_dotenv()
os.environ["OPENAI_BASE_URL"] = os.getenv("baseurl")
os.environ["OPENAI_API_KEY"] = os.getenv("apikey")
model = os.getenv("model")

chat_model = ChatOpenAI(model=model)

system_message = SystemMessage(content="你是一个数学家")
human_message = HumanMessage(content="1+1等于多少？")
ai_message = AIMessage(content="2")
custom_message = ChatMessage(role="user", content="1+2等于多少？")

messages = [system_message, human_message, ai_message, custom_message]
print(messages)

result = chat_model.invoke(messages)
print(f"大模型输出结果：{result.content}")

输出示例：

[SystemMessage(content='你是一个数学家', additional_kwargs={}, response_metadata={}), HumanMessage(content='1+1等于多少？', additional_kwargs={}, response_metadata={}), AIMessage(content='2', additional_kwargs={}, response_metadata={}), ChatMessage(content='1+2等于多少？', additional_kwargs={}, response_metadata={}, role='user')]
大模型输出结果：3

5、多轮对话上下文记忆

现实中很多智能助手、聊天机器人、代码/脚本生成任务，都需要“记住”历史上下文，实现多轮对话和增量推理。

原理：把每次对话内容放入一个列表中，连同新内容一起发给大模型实现。

示例1：无上下文记忆

chat_model = ChatOpenAI(model=model)
result = chat_model.invoke("你叫什么名字?")
print(f"大模型输出结果：{result.content}")

输出结果：

大模型输出结果：你好！我是DeepSeek，由深度求索公司创造的AI助手。很高兴认识你！😊

示例2：有上下文记忆

system_message = SystemMessage(content="你是一个数学家，你叫小王")
human_message = HumanMessage(content="你叫什么名字")
result = chat_model.invoke([system_message, human_message])
print(f"大模型输出结果：{result.content}")

输出结果：

大模型输出结果：我叫小王，是一名数学家。有什么数学问题需要我帮忙解答吗？

6、模型调用输出

6.1 流式输出

流式输出：一种 更具交互感 的模型输出方式，用户不再需要等待完整答案，而是能看到模型 逐个 token 地实时返回内容。

举例：用户提问，请编写一首诗，当问题刚刚发送，系统就开始 一字一句 （逐个token）进行回复，感觉是一边思考一边输出。

更像是“实时对话”，更为贴近人类交互的习惯，更有吸引力。适合构建强调“实时反馈”的应用，如聊天机器人、写作助手等。Langchain 中通过设置 streaming=True 并配合回调机制来启用流式输出。

chat_model = ChatOpenAI(model=model, streaming=True)

result = chat_model.stream("hello world")
for chunk in result:
    print(chunk.content, end="", flush=True)

6.2 非流式输出

非流式输出：这是Langchain与LLM交互时的 默认行为 ，是最简单、最稳定的语言模型调用方式。当用户发出请求后，系统在后台等待模型 生成完整响应 ，然后 一次性将全部结果返回 。

举例：用户提问，请编写一首诗，系统在静默数秒后突然弹出了完整的诗歌。（体验较单调）

在大多数问答、摘要、信息抽取类任务中，非流式输出提供了结构清晰、逻辑完整的结果，适合快速集成和部署。

chat_model = ChatOpenAI(model=model)

result = chat_model.stream("请帮我写一首诗")
print(result)

6.3 批量调用

Batch 批量调用是指一次性向模型发送多个输入，然后批量获取多个输出的处理方式。与逐个调用相比，批量调用可以显著提高处理效率，特别是在处理大量数据时。

messages_list = [
    [HumanMessage(content="用一句话描述太阳")],
    [HumanMessage(content="用一句话描述大海")],
    [HumanMessage(content="用一句话描述星星")]
]

results = chat_model.batch(messages_list)
for result in results:
    print(result.content)

输出示例：

太阳是太阳系中心一颗炽热的气球体，通过核聚变源源不断地释放光与热。
大海是地球蔚蓝的心脏，涌动着无尽的呼吸与生命的律动。
夜幕上钻石般的眼睛，无声地讲述着永恒的故事。

6.4 异步批量调用（了解）

• astream : 异步流式响应ainvoke : 异步处理单条输入

• abatch : 异步处理批量输入

• astream_log : 异步流式返回中间步骤，以及最终响应

• astream_events : （测试版）异步流式返回链中发生的事件（在 langchain-core 0.1.14 中引入）

messages_list = [
    [HumanMessage(content="用一句话描述太阳")],
    [HumanMessage(content="用一句话描述大海")]
]


async def main():
    results = await chat_model.abatch(messages_list)
    for result in results:
        print(result.content)


# 运行异步主函数
asyncio.run(main())