🔥 SystemMessage ：设定AI行为规则或背景信息。比如设定AI的初始状态、行为模式或对话的总

体目标。比如“作为一个代码专家”，或者“返回json格式”。通常作为输入消息序列中的第一个

传递。

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

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

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

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

注意：

FunctionMessage和ToolMessage分别是在函数调⽤和⼯具调⽤场景下才会使⽤的特殊消息类

型，HumanMessage、AIMessage和SystemMessage才是最常⽤的消息类型。

from langchain_core.messages import HumanMessage, SystemMessage

system_message = SystemMessage(content="你是一个英语教学方面的专家")
human_message = HumanMessage(content="帮我制定一个英语六级学习的计划")

messages = [system_message, human_message]

print(messages)

from langchain_core.messages import HumanMessage, SystemMessage

system_message = SystemMessage(content="你是一个英语教学方面的专家",
                               additional_kwargs={"tool":"invoke_dunc1"})
human_message = HumanMessage(content="帮我制定一个英语六级学习的计划")

messages = [system_message, human_message]

print(messages)

from langchain_core.messages import (
AIMessage,
HumanMessage,
SystemMessage,
ChatMessage
)
# 创建不同类型的消息
system_message = SystemMessage(content="你是一个专业的数据科学家")
human_message = HumanMessage(content="解释一下随机森林算法")
ai_message = AIMessage(content="随机森林是一种集成学习方法...")
custom_message = ChatMessage(role="analyst", content="补充一点关于超参数调优的信息")
print(system_message.content)
print(human_message.content)
print(ai_message.content)
print(custom_message.content)

为了尽可能简化自定义链的创建，我们实现了一个"Runnable"协议。许多LangChain组件实现了

Runnable 协议，包括聊天模型、提示词模板、输出解析器、检索器、代理(智能体)等。

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

非流式输出:等待需要的数据全部响应完成后,一次性全部呈现出来

流式输出:边响应边输出(目前市面上的大多数大模型都是如此)

非流式输出：体会invoke()阻塞式的调用

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)

非流式输出:体会流式输出stream()

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流式输出结束")

批量调用:使用batch批量调用

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_KEY1")
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)

异步调用:ainvoke()

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}秒")

PromptTemplate的使用

PromptTemplate如何获取实例

方式1:使用构造方法的方式

from langchain_core.prompts import PromptTemplate
#定义多变量模板
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_core.prompts import PromptTemplate

# 1.创建PromptTemplate的实例
prompt_template = PromptTemplate.from_template(template="你是一个{role},你的名字叫{name}")

prompt = prompt_template.format(role="人工智能专家",name="小智")

print(prompt)

两种特殊结构的使用(部分提示词模板的使用,组合提示词的使用)

部分提示词模板的使用(重点)

方式1:在prompt_template的构造方法或from_template()方法内,使用partial_variables设置

from langchain_core.prompts import PromptTemplate

#定义多变量模板
template = PromptTemplate(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。",
    partial_variables={
        "aspect1":"电池续航", # 设置默认值
        "aspect2":"拍照质量" # 设置默认值
    })
#使用模板生成提示词
prompt_1 = template.format(product="智能手机", aspect1="电池续航11", aspect2="拍照质量22")
print("提示词1:", prompt_1)

方式2:调用方法partial()

from langchain_core.prompts import PromptTemplate

#定义多变量模板
template = PromptTemplate(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。",
    input_variables=["product", "aspect1", "aspect2"],
    ).partial(aspect1="电池续航",aspect2="拍照质量",inplace=True)

#使用模板生成提示词
prompt_1 = template.format(product="智能手机")
print("提示词1:", prompt_1)

组合提示词(了解)

from langchain_core.prompts import PromptTemplate
template = (
PromptTemplate.from_template("Tell me a joke about {topic}")
+ ", make it funny"
+ "\n\nand in {language}"
)
prompt = template.format(topic="sports", language="spanish")
print(prompt)

给变量赋值的两种方式:format()/invoke()

format(): 参数部分:给变量赋值; 返回值:str类型

from langchain_core.prompts import PromptTemplate
#定义多变量模板
template = PromptTemplate.from_template(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。")
#使用模板生成提示词
prompt_1 = template.format(product="笔记本电脑", aspect1="处理速度", aspect2="便携性")

print("提示词1:",prompt_1)
# print(type(prompt_1))

invoke(): 参数部分:使用的是字典; 返回值:PromptValue类型  --推荐

from langchain_core.prompts import PromptTemplate
#定义多变量模板
template = PromptTemplate.from_template(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。")
#使用模板生成提示词
prompt_1 = template.invoke(input={"product":"笔记本电脑","aspect1":"处理速度","aspect2":"便捷性"})

print("提示词1:",prompt_1)
print(type(prompt_1))

结合大模型的使用

import os
import dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

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

os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY1")
# 调用对话模型
chat_model = ChatOpenAI(
    # 必须要设置的三个参数
    model_name='gpt-4o-mini', # 默认使用的是gpt-3.5-turbo模型
    # 当没有显示的声明base_url和api_key的时候,默认会从环境变量中查找
)


# 生成提示词模板
template = PromptTemplate.from_template(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。")

# 给模板的变量赋值
prompt = template.invoke(input={"product":"笔记本电脑","aspect1":"处理速度","aspect2":"便捷性"})

# 调用模型,将提示词传入
response = chat_model.invoke(prompt)

# 查看响应的文本
print(response.content)