LangChain杂记（python版本）-编程实验室

1.连接大模型

先在项目根目录下面建".env"文件，用于存放一些环境变量
连接大模型首先要配置好自己的api key和厂商的base url，将这些信息放在.env里面，便于统一管理，也便于后期上传代码到远程仓库时忽略这个文件，从而防止key泄露
```
DEEPSEEK_API_KEY=sk-............. DEEPSEEK_BASE_URL=https://api.xxxxx.com
```

使用init_chat_model依赖来连接大模型

import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model load_dotenv(verbose=True) #从环境变量读取配置 DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY") DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL") model = init_chat_model( model="大模型名", api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL, )

2.模型调用方式

invoke() ：阻塞式，一次性返回完整结果问答、批处理任务、无需实时反馈的场景。
ainvoke() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。

stream() ：流式输出，实时返回每个token聊天机器人、长文本生成、需要提升用户体验的交互应用。

for chunk in model.stream("写一首七言律诗，总结大模型的发展"): print(chunk.text, end="", flush=True) # 逐token输出

asteam() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。
batch() ：批量处理多个输入高并发场景，需要同时处理大量请求。
abatch() ：非阻塞式，提高系统吞吐量高并发Web应用、IO密集型任务。

3.消息对象字段说明

SystemMessage("你是一个善解人意的助手") HumanMessage("你好！") AIMessage（"你好！"） #ToolMessage的使用 messages = [ HumanMessage(content="北京天气如何"), AIMessage( content="", tool_calls=[ { "name": "get_weather", "args": {"location": "北京"}, #必须和ToolMessage里面的tool_call_id一致 "id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s" } ] ), ToolMessage( content="今天北京天气晴朗，万里无云~", tool_call_id="call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s" ) ]

4.格式化输出AI的回复信息

#1. `SystemMessage`：设定角色，告诉模型你叫小智 #2. `HumanMessage`：用户提问 #3. `model.invoke(messages)`：发给智谱 API 拿到返回 #4. `rprint`：rich 库美化格式化输出，看得更清晰 #5. `response.response_metadata`：打印 token 消耗、请求元信息 from langchain_core.messages import SystemMessage, HumanMessage from rich import print as rprint # rich库美化打印输出 messages = [ SystemMessage("你叫小智，是一名助人为乐的助手。"), HumanMessage("你好，好久不见，请介绍下你自己。") ] response = model.invoke(messages) rprint(response) print(response.response_metadata)

5.对话历史优化（保存记忆）

记住记忆（问题：对话历史会越来越长，消耗大量的token和成本）

conversation = [] # 第一次 conversation.append({"role": "user", "content": "我叫张三"}) response1 = model.invoke(conversation) # 关键：保存 AI 回复 conversation.append({"role": "assistant", "content": response1.content}) # 第二次（传递完整历史） conversation.append({"role": "user", "content": "我叫什么？"}) response2 = model.invoke(conversation) # AI 记得！

对话历史优化（只保留最近N轮对话）

总是保留system消息（定义角色）

只保留最近N轮对话，丢弃更早的历史

#初始化 long_conversation=[{"role":"system","content":"你是python导师"}] #第一轮对话 long_conversation.append({"role":"user","content":"什么是列表？用一句话解释"}) r1=model.invoke(long_conversation) long_conversation.append({"role":"assistant","content":r1.content}) # 第 2 轮 long_conversation.append({"role": "user", "content": "列表和元组有什么区别？用一句解释"}) r2 = model.invoke(long_conversation) long_conversation.append({"role": "assistant", "content": r2.content}) # 第 3 轮 long_conversation.append({"role": "user", "content": "什么是字典呢？用一句解释"}) r3 = model.invoke(long_conversation) long_conversation.append({"role": "assistant", "content": r3.content}) print(f"原始消息数: {len(long_conversation)}") def keep_recent_messages(messages, max_pairs=3): """ 保留最近的 N 轮对话 max_pairs: 保留的对话轮数（每轮 = user + assistant） """ # 分离 system 和对话 system_msgs = [m for m in messages if m.get("role") == "system"] conversation_msgs = [m for m in messages if m.get("role") != "system"] # 只保留最近的 `[ -N : ]`：取列表**倒数 N 个元素**，从倒数第 N 个一直到列表末尾 #一轮对话包含了user和assistant，所以要*2 recent_msgs = conversation_msgs[-(max_pairs * 2):] # 返回：system + 最近对话 return system_msgs + recent_msgs # 优化：只保留最近 2 轮 optimized = keep_recent_messages(long_conversation, max_pairs=2) print(f"优化后消息数: {len(optimized)}") print(f"保留的内容: system + 最近2轮对话")

6.借助 content_blocks ，我们可以用一套标准代码，无缝地在不同厂商的模型之间切换

content_blocks=[ {'type': 'text', 'text': '这张图里有什么？'}, {'type': 'image','base64': base64_image,'mime_type': 'image/png',} ]

7.ChatPromptTemplate 是用于生成消息列表的核心组件

调用：
- invoke()，传参是字典列表；返回值类型是ChatPromptValue
- format()，参数是变量值，返回值是字符串
- format_message()，传入的参数是变量值；发挥类型是消息列表

8.提示词模板的高级特性

部分变量预填充：partial()。预填充某些固定不变的变量，创建模板的变体。

原始编写的提示词：

template=ChatPromptTemplate.from_messages([ ("system","你是{role},目标用户是{audience}"), ("user","{task}") ]) result1=template.invoke({"role":"导游","audience":"游客","task":"介绍一下故宫"}) result2=template.invoke({"role":"导游","audience":"游客","task":"介绍一下圆明园"}) print(result1) print(result2)

使用partial()后：

template=ChatPromptTemplate.from_messages([ ("system","你是{role},目标用户是{audience}"), ("user","{task}") ]) customer_support_template=template.partial(role="导游",audience="游客") result1=customer_support_template.invoke({"task":"介绍一下故宫"}) result2=customer_support_template.invoke({"task":"介绍一下圆明园"}) print(result1) print(result2)