Mac 使用transformers 库运行Qwen3-Embedding-0.6B模型使用全记录
Mac 使用transformers 库运行本地大模型使用全记录下载大模型到本地//创建本地运行虚拟环境 conda create --name vllm_dev python=3.12 conda activate vllm_dev //刚开始想用 vllm 运行模型，但是发现这个模型很小，改用transformers库直接运行，更省资源 //huggingface下载Qwen3-Embedding-0.6B模型 pip install huggingface_hub export HF_ENDPOINT=https://hf-mirror.com hf download Qwen/Qwen3-Embedding-0.6B >Fetching 12 files: 100%|████████████████████████████████████████████| 12/12 [01:14<00:00, 6.20s/it] Download complete: : 1.21GB [01:14, 19.4MB/s] /Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418 本地运行Qwen3-Embedding-0.6B模型//安装相关包pip install torch transformers fastapi uvicorn tiktoken sentencepiece //包解释torch —— 模型推理 transformers —— 加载 Qwen3 模型 fastapi —— API 服务 uvicorn —— 运行服务器 tiktoken —— Qwen3 分词器必须 sentencepiece —— 兼容依赖 开启本地服务emb_serve.py 启动服务：❯ python emb_server.py 🚀 当前计算设备: MPS 📥 正在加载模型: /Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418 ... `torch_dtype` is deprecated! Use `dtype` instead! Loading weights: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 310/310 [00:00<00:00, 359.07it/s] ✅ 模型加载完成！服务准备就绪。 INFO: Started server process [7296] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8001 (Press CTRL+C to quit) INFO: 127.0.0.1:53982 - "GET /docs HTTP/1.1" 200 OK INFO: 127.0.0.1:53983 - "GET /redoc HTTP/1.1" 200 OK INFO: 127.0.0.1:54022 - "GET /openapi.json HTTP/1.1" 200 OK INFO: 127.0.0.1:54121 - "POST /v1/embeddings HTTP/1.1" 200 OK emb_server.pyimport torch import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Union, Optional from transformers import AutoTokenizer, AutoModel # ========================================== # 1. 配置区域 (请在此处修改本地模型路径) # ========================================== # 【关键】如果你的模型在本地，请直接填入绝对路径 # 例如: "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/..." # 如果留空或者填英文名，它会尝试去联网下载 MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" # 设备检测：优先使用 Mac 的 MPS (Metal Performance Shaders) DEVICE = "mps" if torch.backends.mps.is_available() else "cpu" print(f"🚀 当前计算设备: {DEVICE.upper()}") # ========================================== # 2. 初始化 FastAPI 应用 # ========================================== app = FastAPI( title="Qwen3-Embedding-0.6B API", description="本地部署的 Embedding 服务，运行于 Mac M-Series 芯片", swagger_js_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/5.11.0/swagger-ui-bundle.min.js", swagger_css_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/5.11.0/swagger-ui.min.css", version="1.0.0" ) # ========================================== # 3. 加载模型 (全局单例) # ========================================== print(f"📥 正在加载模型: {MODEL_PATH} ...") try: # 加载分词器 tokenizer = AutoTokenizer.from_pretrained( MODEL_PATH, trust_remote_code=True, local_files_only=True # 强制只读取本地，防止联网超时 ) # 加载模型 # torch_dtype=torch.float16 适合 Mac M 系列芯片，显存占用更低 model = AutoModel.from_pretrained( MODEL_PATH, trust_remote_code=True, local_files_only=True, torch_dtype=torch.float16 ).to(DEVICE) # 设置为评估模式 model.eval() print("✅ 模型加载完成！服务准备就绪。") except Exception as e: print(f"❌ 模型加载失败: {e}") print("💡 请检查 MODEL_PATH 路径是否正确，以及是否安装了 transformers 和 torch。") # 即使加载失败也不退出进程，方便你调试，但接口会不可用 model = None tokenizer = None # ========================================== # 4. 数据模型定义 (Pydantic) # ========================================== class EmbeddingRequest(BaseModel): input: Union[str, List[str]] # 支持单个字符串或字符串列表 model: Optional[str] = None # 兼容 OpenAI 格式，虽然这里不用 class EmbeddingData(BaseModel): object: str = "embedding" embedding: List[float] index: int class EmbeddingResponse(BaseModel): object: str = "list" data: List[EmbeddingData] model: str # ========================================== # 5. 核心接口 # ========================================== @app.post("/v1/embeddings", response_model=EmbeddingResponse) async def get_embeddings(request: EmbeddingRequest): # 检查模型是否加载成功 if model is None: raise HTTPException(status_code=503, detail="模型尚未加载完成，请稍后再试。") # 1. 预处理输入 inputs = request.input if isinstance(inputs, str): inputs = [inputs] if not inputs: raise HTTPException(status_code=400, detail="输入不能为空") # 2. 推理过程 (使用 torch.no_grad() 节省显存) with torch.no_grad(): # 分词 # max_length=8192 是 Qwen3 的上下文窗口大小 encoded_input = tokenizer( inputs, padding=True, truncation=True, return_tensors='pt', max_length=8192 ).to(DEVICE) # 模型前向传播 # 输出包含 last_hidden_state (所有token的向量) model_output = model(**encoded_input) # 3. 平均池化 # 这一步是为了把一句话里所有字的向量聚合成一个代表整句话的向量 input_mask_expanded = encoded_input['attention_mask'].unsqueeze(-1).expand( model_output.last_hidden_state.size()).float() sentence_embeddings = torch.sum(model_output.last_hidden_state * input_mask_expanded, 1) / torch.clamp( input_mask_expanded.sum(1), min=1e-9) # 4. 归一化 # Embedding 向量通常需要 L2 归一化，这样计算余弦相似度时才准确 sentence_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1) # 5. 转回 CPU 并转为列表 # MPS 张量不能直接转 numpy，必须先 .cpu() embeddings = sentence_embeddings.cpu().numpy().tolist() # 6. 构造返回结果 data = [ EmbeddingData(embedding=emb, index=i) for i, emb in enumerate(embeddings) ] return EmbeddingResponse(data=data, model="Qwen3-Embedding-0.6B") @app.get("/") def read_root(): return { "status": "running", "device": DEVICE, "message": "Qwen3-Embedding-0.6B API 已启动", "docs": "/docs" } # ========================================== # 6. 启动入口 # ========================================== if __name__ == "__main__": # host="0.0.0.0" 允许局域网访问 # port=8001 指定端口 uvicorn.run(app, host="0.0.0.0", port=8001) 也可以不用启动服务，直接访问本地的 Embeddingfrom langchain_huggingface import HuggingFaceEmbeddings MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" embeddings = HuggingFaceEmbeddings( model_name=MODEL_PATH, model_kwargs={"device": "mps"}, encode_kwargs={"normalize_embeddings": True} ) 本地向量化代码测试# 本地Embedding Models测试 import requests import json # 1. 定义请求地址 (确保端口是 8001) url = "http://127.0.0.1:8001/v1/embeddings" # 2. 准备数据 # 注意：这里兼容 OpenAI 格式，所以加了 "model" 字段，虽然服务端可能不校验 payload = { "model": "Qwen3-Embedding-0.6B", "input": "你好，这是一个测试文本。" } headers = { "Content-Type": "application/json" } # 3. 发送 POST 请求 try: response = requests.post(url, headers=headers, json=payload) # 检查状态码 if response.status_code == 200: data = response.json() # 提取向量 embedding_vector = data['data'][0]['embedding'] print(f"✅ 请求成功！") print(f"模型: {data.get('model')}") print(f"向量维度: {len(embedding_vector)}") print(f"向量前10个数值: {embedding_vector[:10]}") else: print(f"❌ 请求失败: {response.status_code}") print(response.text) except requests.exceptions.ConnectionError: print("❌ 连接错误：无法连接到 127.0.0.1:8001") print("请确保你的 Mac 本地终端中正在运行 app.py 且没有报错。") 运行结果：✅ 请求成功！ 模型: Qwen3-Embedding-0.6B 向量维度: 1024 向量前10个数值: [-0.02349964901804924, -0.1068049818277359, -0.014799184165894985, -0.05713353306055069, 0.016370529308915138, 0.08789951354265213, -0.024308187887072563, 0.0869838148355484, -0.07337475568056107, 0.04606372490525246] 在 langchain 中使用import os from langchain_openai import OpenAIEmbeddings API_BASE_URL = "http://127.0.0.1:8001/v1" embeddings = OpenAIEmbeddings( model="Qwen3-Embedding-0.6B", # 这里填你在 FastAPI 中定义的 model 名称 base_url=API_BASE_URL, api_key="sk-no-key-required", check_embedding_ctx_length=False # 关闭长度检查，因为本地模型可能有自己的截断逻辑 ) try: text = "你好，这是一个测试文本。" print(f"正在对文本进行向量化: '{text}' ...") # 调用接口获取向量 query_result = embeddings.embed_query(text) print(f"✅ 成功！") print(f"向量维度: {len(query_result)}") print(f"前5个数值: {query_result[:5]}") except Exception as e: print(f"❌ 发生错误: {e}") print("提示：请检查 Mac 终端上的 FastAPI 服务是否正在运行。") //运行结果正在对文本进行向量化: '你好，这是一个测试文本。' ... ✅ 成功！ 向量维度: 1024 前5个数值: [-0.02349964901804924, -0.1068049818277359, -0.014799184165894985, -0.05713353306055069, 0.016370529308915138]
加载TXT的文件#安装langchain_community # pip install langchain_community # 读取普通的文件 from langchain_community.document_loaders import TextLoader text = TextLoader(file_path="asset/load/01-langchain-utf-8.txt",encoding="utf-8") docs = text.load() print(docs) # Documment对象中有两个重要的属性： # page_content：真正的文档内容 # metadata：文档内容的原数据 加载pdf#加载pdf #LangChain加载PDF文件使用的是pypdf # pip install pypdf from langchain_community.document_loaders import PyPDFLoader pdfLoader = PyPDFLoader(file_path="asset/load/02-load.pdf") docs = pdfLoader.load() # print(type(docs)) # print(docs[0]) for doc in docs: print(f"文档的内容：{doc.page_content}") # type(pages[0]) #langchain_core.documents.base.Document # pages[0].page_content #只获取本页内容 # pages[0].metadata # {...,'source': './asset/load/load.pdf',.., 'page': 0} load_and_split 加载并切分from langchain_community.document_loaders import PyPDFLoader pdfLoader = PyPDFLoader(file_path="asset/load/02-load.pdf") docs = pdfLoader.load_and_split() # print(docs) for doc in docs: print(f"❀{doc.page_content}") 加载CSVfrom langchain_community.document_loaders import CSVLoader loader = CSVLoader(file_path="asset/load/03-load.csv") data= loader.load() print(type(data[0].page_content)) #打印每一行 for doc in data: print(f"{doc.page_content}\n") 加载JSON#安装jq pip install jq from langchain_community.document_loaders import JSONLoader #情况1 打印完整json数据 loader = JSONLoader( file_path="asset/load/04-load.json", jq_schema=".", text_content=False # <--- 关键修改：允许 page_content 是 dict 类型 ) docs = loader.load() print(docs[0].page_content) # 情况2 # .messages[].content:遍历.messages[]中所有元素 从每一个元素中提取.content字段 json_loader = JSONLoader( file_path="asset/load/04-load.json", jq_schema=".messages[].content", ) # 3.加载 doc2 = json_loader.load() print("-" * 60) print(doc2[0].page_content) 提取JSON内容#json 示例2 提取04-response.json文件中嵌套在 data.items[].content 的文本 # 如果希望处理 JSON 中的 嵌套字段、数组元素提取，可以使用 content_key 配合is_content_key_jq_parsable=True ，通过 jq 语法精准定位目标数据。 # 通常，对api请求结果的采集 # "status": "success", # "data": { # "page": 2, # "per_page": 3, # "total_pages": 5, # "total_items": 14, # "items": [ # { # "id": 101, # "title": "Understanding JSONLoader", # "content": "This article explains how to parse API responses...", # "author": { # "id": "user_1", # "name": "Alice" # }, # "created_at": "2023-10-05T08:12:33Z" # }, from langchain_community.document_loaders import JSONLoader file_path = 'asset/load/04-response.json' # 方式1： loader = JSONLoader( file_path=file_path, jq_schema=".data.items[].content", ) docs = loader.load() for doc in docs: print(f"文档内容:{doc.page_content}") # 方式2： loder2 = JSONLoader( file_path="asset/load/04-response.json", jq_schema=".data.items[]", # 先定位到数组条目 content_key=".content", # 再从条目中提取 content 字段 is_content_key_jq_parsable=True ) docs2 = loder2.load() print("-"*60) for doc in docs2: print(f"文档内容:{doc.page_content}") 深层提取json文本#提取04-response.json文件中嵌套在 data.items[] 里的 title、content 和 其文本 from langchain_community.document_loaders import JSONLoader from pprint import pprint loader = JSONLoader( file_path=file_path, # jq_schema=".data.items[] | {id, author, text: (.title + '\n' + .content)}", jq_schema=".data.items[]", content_key='(.title + "\n\n" + .content)', is_content_key_jq_parsable=True # 用jq解析content_key ) docs = loader.load() pprint(docs[0].page_content) #取出第二层的作者 loader2 = JSONLoader( file_path=file_path, jq_schema=".data.items[]", content_key='.author.name', is_content_key_jq_parsable=True ) docs2 = loader2.load() print("=== 取出第二层的作者 ===") print(docs2[0].page_content) 加载HTML# 加载HTML(了解) # 1.导入相关的依赖 from langchain_community.document_loaders import UnstructuredHTMLLoader # 2.定义UnstructuredHTMLLoader对象 # strategy: # "fast" 解析加载html文件速度是比较快（但可能丢失部分结构或元数据） # "hi_res": (高分辨率解析) 解析精准（速度慢一些） # "ocr_only" 强制使用ocr提取文本，仅仅适用于图像（对HTML无效） # mode ：one of `{'paged', 'elements', 'single'} # "elements" 按语义元素（标题、段落、列表、表格等）拆分成多个独立的小文档 html_loader = UnstructuredHTMLLoader( file_path="asset/load/05-load.html", mode="elements", strategy="fast" ) # 3.加载 docs = html_loader.load() print(len(docs)) # 16 # 4.打印 for doc in docs: print(doc) 加载Markdown# 2.6 加载Markdown(了解) # pip install markdown # pip install unstructured # 1.导入相关的依赖 from langchain_community.document_loaders import UnstructuredMarkdownLoader from pprint import pprint # 2.定义UnstructuredMarkdownLoader对象 md_loader = UnstructuredMarkdownLoader( file_path="asset/load/06-load.md", strategy="fast" ) # 3.加载 docs = md_loader.load() print(len(docs)) # 4.打印 for doc in docs: pprint(doc)
# 核心概念 # 验证和重试 = 确保 LLM 应用的可靠性和数据质量 # 在生产环境中，需要处理三类问题： # 网络错误 - 临时性连接问题（用 with_retry()） # 模型故障 - 主模型不可用（用 with_fallbacks()） # 输出质量 - LLM 输出不符合要求（用验证 + 重试循环） # 基本用法 # 1. with_retry() - 自动重试 # 处理临时性网络错误： import os import sys from langchain.chat_models import init_chat_model from pydantic import BaseModel, Field, ValidationError sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() def demo1(): # 添加重试机制 llm_with_retry = chat_model.with_retry( retry_if_exception_type=(ConnectionError, TimeoutError), wait_exponential_jitter=True, # 指数退避 + 随机抖动 stop_after_attempt=3, # 最多重试 3 次 ) response = llm_with_retry.invoke("你好") # 工作原理： # 第 1 次尝试 → 失败 (ConnectionError) # ↓ 等待 1s # 第 2 次尝试 → 失败 (ConnectionError) # ↓ 等待 2s # 第 3 次尝试 → 成功 ✓ # 2. with_fallbacks() - 降级方案 # 主模型失败时切换到备用模型： def demo2(): # 主模型 primary_model = init_chat_model("groq:llama-3.3-70b-versatile") # 备用模型（更可靠/更便宜） fallback_model = init_chat_model("groq:llama-3.1-8b-instant") # 配置降级 llm_with_fallbacks = primary_model.with_fallbacks([fallback_model]) response = llm_with_fallbacks.invoke("介绍 Python") # 主模型正常 → 使用主模型 # 主模型失败 → 自动切换到备用模型 # 3. Pydantic 验证 # 使用 Pydantic 约束确保数据质量： def demo3(): from pydantic import BaseModel, Field, field_validator class User(BaseModel): name: str = Field(min_length=2, max_length=20) age: int = Field(ge=0, le=150) # 0-150 岁 email: str @field_validator("email") @classmethod def validate_email(cls, v): if "@" not in v: raise ValueError("邮箱必须包含 @") return v # 使用 try: user = User(name="张三", age=200, email="invalid") # 失败 except ValidationError as e: print(e.errors()) # 查看错误详情 # 核心组件 # 1. with_retry() 参数 # 参数 说明 默认值 # retry_if_exception_type 哪些异常会触发重试 (Exception,) # stop_after_attempt 最大重试次数 3 # wait_exponential_jitter 是否使用指数退避 + 抖动 False # 常见重试异常： # # 网络相关 # (ConnectionError, TimeoutError, httpx.ConnectError) # # API 限流 # (RateLimitError, ) # # 所有异常（谨慎使用） # (Exception, ) # 2. with_fallbacks() 参数 # primary_model.with_fallbacks( # fallbacks=[model1, model2], # 备用模型列表（按顺序尝试） # exceptions_to_handle=(Exception,) # 触发降级的异常 # ) # 工作流程： # 尝试 primary_model → 失败 # ↓ # 尝试 model1 → 失败 # ↓ # 尝试 model2 → 成功 ✓ # 3. Pydantic Field 约束 # 约束 用途 示例 # gt / ge 数值 > / >= Field(gt=0) # lt / le 数值 < / <= Field(le=100) # min_length / max_length 字符串长度 Field(min_length=2) # pattern 正则表达式 Field(pattern=r'^\d{11}$') # class Product(BaseModel): # name: str = Field(min_length=2, max_length=50) # price: float = Field(gt=0, description="价格必须 > 0") # stock: int = Field(ge=0, description="库存 >= 0") # 实际应用 # 1. LLM 输出验证 + 重试循环 # 当 LLM 输出不符合验证规则时，重新提示： def demo4(): from pydantic import ValidationError class Product(BaseModel): name: str = Field(min_length=2) price: float = Field(gt=0) structured_llm = chat_model.with_structured_output(Product) max_retries = 3 text = "产品价格是 -100 元" # 负价格会触发验证失败 for attempt in range(1, max_retries + 1): try: result = structured_llm.invoke(f"提取产品信息：{text}") # 验证通过 break except ValidationError as e: error_msg = e.errors()[0]["msg"] # 在提示中加入错误信息 text = f"{text}\n注意: {error_msg}。请确保价格 > 0" if attempt == max_retries: raise # 重试次数用完 # 第 1 次尝试: # LLM 输出: price = -100 # 验证: ✗ 失败 (price 必须 > 0) # 第 2 次尝试（修正提示）: # 提示: "...注意: price 必须 > 0" # LLM 输出: price = 100 # 验证: ✓ 通过
# 核心概念 # Structured Output = 将 LLM 的自然语言输出转为结构化 Python 对象 # 在 LangChain 1.0 中，使用 with_structured_output() 方法结合 Pydantic 模型，可以确保 LLM 返回符合预定义模式的数据。 # 基本用法 # 定义 Pydantic 模型 from typing import List, Optional from pydantic import BaseModel, Field, model_serializer class Person(BaseModel): # 变量名是英文 (name)，描述可以是中文，帮助模型理解 name: str = Field(description="人的姓名") # 变量名是英文 (age) age: int = Field(description="人的年龄") # 变量名是英文 (job) job: Optional[str] = Field(default=None, description="人的职业") # 自定义序列化方法，消除警告 @model_serializer def serialize_model(self): return {"name": self.name, "age": self.age, "job": self.job} # email: Optional[str] = Field(default=None, description="人的邮箱") import os import sys from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.tools import tool from langchain.agents.middleware import AgentMiddleware from langgraph.checkpoint.memory import InMemorySaver sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() # 使用 with_structured_output() # 核心组件 # 1. Pydantic BaseModel # 所有结构化输出的数据模型都必须继承 BaseModel： # from pydantic import BaseModel # class MyModel(BaseModel): # field1: str # field2: int # 2. Field 描述 # 使用 Field() 添加字段描述，帮助 LLM 理解： from pydantic import Field # 高级特性 # 可选字段 class Person(BaseModel): name: str age: Optional[int] = None job: Optional[str] = None # 可以为 None def demo1(): # 创建结构化输出的 LLM structured_llm = chat_model.with_structured_output(Person, include_raw=False) response = structured_llm.invoke( "张三是一名 30 岁的软件工程师。请提取信息。按照要求返回JSON格式数据，字段名必须为英文（name/age/job）" ) # print(response) print(f"name={response.name} age={response.age} job={response.job}") # 工作原理 # 传统方式 vs 结构化输出 # 传统方式（繁琐）： # 1. 提示词要求 JSON # prompt = "以JSON格式返回：{name, age, occupation}" # response = chat_model.invoke(prompt) # # 2. 手动解析 # import json # data = json.loads(response.content) # # 3. 手动验证类型 # if not isinstance(data["age"], int): # raise ValueError("age must be int") # # 4. 手动创建对象 # person = Person(**data) # 结构化输出（简洁）： # 一步到位 # structured_llm = chat_model.with_structured_output(Person) # person = structured_llm.invoke("张三是一名 30 岁的软件工程师") # ✅ 自动解析、验证、创建对象 # 幕后流程 # 1. Pydantic 模型 → JSON Schema # Person → { # "type": "object", # "properties": { # "name": {"type": "string", "description": "姓名"}, # "age": {"type": "integer", "description": "年龄"} # } # } # 2. JSON Schema → LLM (函数调用) # LLM 被强制返回符合 schema 的 JSON # 3. JSON → Pydantic 对象 # 自动验证类型并创建 Person 实例 # 创建结构化输出的 LLM # 实际应用 # 1. 客户信息提取 class CustomerInfo(BaseModel): name: str = Field(description="客户姓名") phone: str = Field(description="电话号码") email: Optional[str] = Field(None, description="邮箱") issue: str = Field(description="问题描述") def demo2(): structured_llm = chat_model.with_structured_output(CustomerInfo) conversation = "客户: 我是李明，电话 138-1234-5678，问题是：订单没发货.仅返回JSON数据，不添加任何额外说明文字" info = structured_llm.invoke(f"提取客户信息：{conversation}") print(info) # return self.__pydantic_serializer__.to_python( # name='李明' phone='138-1234-5678' email=None issue='订单没发货' # 应用： # 自动填充 CRM 系统 # 工单自动分类 # 客服辅助 # 2. 产品评论分析 class Review(BaseModel): product: str rating: int = Field(description="评分 1-5") pros: List[str] = Field(description="优点列表") cons: List[str] = Field(description="缺点列表") def demo3(): structured_llm = chat_model.with_structured_output(Review) review = structured_llm.invoke( """ iPhone 15 很棒！摄像头强大，手感好。但是价格贵，没有充电器。4分。仅返回JSON数据，不添加任何额外说明文字 """ ) print(review) # return self.__pydantic_serializer__.to_python( # product='iPhone 15' rating=4 pros=['摄像头强大', '手感好'] cons=['价格贵', '没有充电器'] # 应用： # 批量处理用户评论 # 自动生成分析报告 # 发现产品改进点 # 3. 文档信息提取 嵌套模型 # 1. 定义子模型：商品明细 class InvoiceItem(BaseModel): name: str = Field(description="商品名称") amount: float = Field(description="商品单价或该项小计金额") # quantity: int = Field(default=1, description="数量，如果未提及默认为1") # 可选 # 2. 定义主模型：发票 class Invoice(BaseModel): invoice_number: str = Field(description="发票号") date: str = Field(description="日期，格式务必为 YYYY-MM-DD") total_amount: float = Field(description="发票总金额") items: List[InvoiceItem] = Field(description="商品明细列表，包含名称和价格") def demo4(): structured_llm = chat_model.with_structured_output(Invoice) document = """ 发票号：20230801001 日期：2023-08-01 总金额：1234.56 项目： 1. 商品A - 100.00 2. 商品B - 200.00 3. 服务费 - 934.56 """ # 优化后的 Prompt：更自然，不需要强调“仅返回JSON”，因为 structured_output 已经做了约束 prompt_text = f""" 请从以下文本中提取发票信息，并**以 JSON 格式返回**结果： {document} 注意： 1. 确保日期格式为 YYYY-MM-DD。 2. 将每个项目拆分为名称和对应的金额。 3. 如果总金额与各项之和不符，以文档中明确写出的"总金额"为准。 """ info: Invoice = structured_llm.invoke(prompt_text) print(info) # 应用： # 自动化财务处理 # OCR 后结构化 # 数据录入 # 最佳实践 # # 1. 使用清晰的字段描述 # class Good(BaseModel): # created_at: str = Field(description="创建时间，格式 YYYY-MM-DD") # class Bad(BaseModel): # created_at: str # 没有描述，LLM 可能格式错误 # # 2. 合理使用 Optional # class Good(BaseModel): # email: Optional[str] = None # 邮箱可能没有 # class Bad(BaseModel): # email: str # 强制必填，可能导致提取失败 # # 3. 使用枚举限制值 # class Good(BaseModel): # status: Status # 枚举 # class Bad(BaseModel): # status: str # 可能返回任意字符串 # # 4. 列表设置合理的描述 # class Good(BaseModel): # tags: List[str] = Field(description="产品标签，如 '电子产品', '手机'") # class Bad(BaseModel): # tags: List[str] # LLM 不知道该提取什么 # # 5. 嵌套模型保持简单 # class Good(BaseModel): # user: User # 1 层嵌套 # settings: dict # 复杂数据用 dict # class Bad(BaseModel): # user: User # company: Company # address: Address # country: Country # 4 层嵌套，容易出错 if __name__ == "__main__": # demo1() # demo2() # demo3() demo4()
""" LangChain 1.0 - Middleware Basics (中间件基础) ============================================== 本模块重点讲解： 1. 什么是中间件（Middleware） 2. before_model 和 after_model 钩子 3. 自定义中间件的创建 4. 多个中间件的组合 5. 内置中间件的使用 """ import os import sys from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.tools import tool from langchain.agents.middleware import AgentMiddleware from langgraph.checkpoint.memory import InMemorySaver sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() @tool def get_weather(city: str) -> str: """ 获取指定城市的天气信息 参数: city (str): 要查询天气的城市名称 例如 "北京" 返回: str: 包含天气信息的字符串，格式为 "城市 天气描述 温度" """ # 模拟天气数据 weather_data = { "北京": "晴天，温度 15°C，空气质量良好", "上海": "多云，温度 18°C，有轻微雾霾", "深圳": "阴天，温度 22°C，可能有小雨", "成都": "小雨，温度 12°C，湿度较高", } # 检查城市是否在数据中 if city not in weather_data: return f"城市 {city} 的天气信息未找到" # 返回天气信息 return weather_data.get(city, f"抱歉，暂时没有{city}的天气数据") # ============================================================================ # 示例 1：最简单的中间件 """ 关键规则 1. 必须继承 AgentMiddleware ← 这个固定 2. 方法名固定 (before_model, after_model) ← 这个固定 3. 类名随意 ← 这个不固定 LangGraph 只看： - 是否继承 AgentMiddleware？ - 是否有 before_model / after_model 方法？ """ import logging from typing import Any, Dict, Optional # 配置 logger logger = logging.getLogger(__name__) # ============================================================================ class LoggingMiddleware(AgentMiddleware): """ 日志中间件 - 记录每次模型调用 """ def before_model( self, state: Dict[str, Any], runtime: Any ) -> Optional[Dict[str, Any]]: """模型调用前""" messages = state.get("messages", []) logger.info(f"[Middleware] Before Model Call | Message Count: {len(messages)}") # 这里可以添加更多上下文信息，例如 runtime.request_id return None # 继续正常流程 def after_model( self, state: Dict[str, Any], runtime: Any ) -> Optional[Dict[str, Any]]: """模型响应后""" messages = state.get("messages", []) if not messages: logger.warning( "[Middleware] After Model Call | No messages found in state." ) return None last_message = messages[-1] msg_type = last_message.__class__.__name__ logger.info(f"[Middleware] After Model Call | Response Type: {msg_type}") # 如果是特定类型的消息，可以提取内容摘要记录 # if hasattr(last_message, 'content'): # logger.debug(f"Response Content Preview: {str(last_message.content)[:100]}...") return None # 不修改状态 # ============================================================================ # 示例 2：修改状态的中间件 # ============================================================================ class CallCounterMiddleware(AgentMiddleware): """ 计数中间件 - 统计模型调用次数 在中间件内部维护计数器（简单版本） """ def __init__(self): super().__init__() self.count = 0 # 简单计数器 def after_model(self, state, runtime): """模型响应后，增加计数""" self.count += 1 print(f"\n[计数器] 模型调用次数: {self.count}") return None # 不修改 state # ============================================================================ # 示例 3：消息修剪中间件 # ============================================================================ class MessageTrimmerMiddleware(AgentMiddleware): """ 消息修剪中间件 - 限制消息数量 before_model 修改消息列表 注意：需要配合无 checkpointer 使用，否则历史会被恢复 """ def __init__(self, max_messages=5): super().__init__() self.max_messages = max_messages self.trimmed_count = 0 # 统计修剪次数 def before_model(self, state, runtime): """模型调用前，修剪消息""" messages = state.get("messages", []) if len(messages) > self.max_messages: # 保留最近的 N 条消息 trimmed_messages = messages[-self.max_messages :] self.trimmed_count += 1 print( f"\n[修剪] 消息从 {len(messages)} 条减少到 {len(trimmed_messages)} 条 (第{self.trimmed_count}次修剪)" ) return {"messages": trimmed_messages} return None # ============================================================================ # 示例 4：输出验证中间件 # ============================================================================ class OutputValidationMiddleware(AgentMiddleware): """ 输出验证中间件 - 检查响应长度 after_model 验证输出 """ def __init__(self, max_length=100): super().__init__() self.max_length = max_length def after_model(self, state, runtime): """模型响应后，验证输出""" messages = state.get("messages", []) if not messages: return None last_message = messages[-1] content = getattr(last_message, "content", "") if len(content) > self.max_length: print( f"\n[警告] 响应���长 ({len(content)} 字符)，已截断到 {self.max_length}" ) # 这里可以实现截断或重试逻辑 return None # ============================================================================ # 示例 5：条件跳转（高级） # ============================================================================ class MaxCallsMiddleware(AgentMiddleware): """ 最大调用限制中间件 通过抛出异常来阻止模型调用（更可靠的方式） """ def __init__(self, max_calls=3): super().__init__() self.max_calls = max_calls self.count = 0 # 简单计数器 def before_model(self, state, runtime): """检查调用次数，超过限制则抛出异常""" if self.count >= self.max_calls: print(f"\n[限制] 已达到最大调用次数 {self.max_calls}，停止调用") # 抛出自定义异常来阻止继续执行 raise ValueError(f"已达到最大调用次数限制: {self.max_calls}") print(f"[限制] 当前调用次数: {self.count}/{self.max_calls}") return None def after_model(self, state, runtime): """增加计数""" self.count += 1 print("次数+1") return None # ============================================================================ # 示例 6：内置中间件使用 # ============================================================================ from langchain.agents.middleware import SummarizationMiddleware def demo1(): """ 示例 1：最简单的中间件 """ # 创建代理时添加中间件 agent = create_agent( model=chat_model, tools=[get_weather], checkpointer=InMemorySaver(), middleware=[ LoggingMiddleware(), CallCounterMiddleware(), MessageTrimmerMiddleware(), SummarizationMiddleware( # 内置中间件 model="groq:llama-3.3-70b-versatile", max_tokens_before_summary=200, # 超过 200 token 就摘要 ), ], # 添加日志中间件 ) # 2. 构造 config 对象 config = {"configurable": {"thread_id": "demo1"}} response = agent.invoke( {"messages": [{"role": "user", "content": "你好，今天北京的天气怎么样"}]}, config, ) print(f"Agent: {response['messages'][-1].content}") # 结果 # [中间件] before_model: 准备调用模型 # [中间件] 当前消息数: 1 # [中间件] after_model: 模型已响应 # [中间件] 响应类型: AIMessage # [中间件] before_model: 准备调用模型 # [中间件] 当前消息数: 3 # [中间件] after_model: 模型已响应 # [中间件] 响应类型: AIMessage # Agent: 今天北京的天气是晴天，温度 15°C，空气质量良好。 if __name__ == "__main__": demo1()
# Middleware（中间件）= Agent 执行过程中的钩子函数 # 在 LangChain 1.0 中，中间件是处理 Agent 生命周期的标准方式。 # 基本用法 # 创建自定义中间件 import sys import os from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver from langchain.agents.middleware import AgentMiddleware chat_model = get_chat_model() class MyMiddleware(AgentMiddleware): def before_model(self, state, runtime): """模型调用前执行""" print("准备调用模型") return None # 返回 None 表示继续正常流程 def after_model(self, state, runtime): """模型响应后执行""" print("模型已响应") return None # 返回 None 表示不修改状态 def demo1(): # 使用中间件 agent = create_agent(model=chat_model, tools=[], middleware=[MyMiddleware()]) result = agent.invoke({"messages": [{"role": "user", "content": "你好"}]}) print(result["messages"][-1].content) # 准备调用模型 # 模型已响应 # 你好！👋 很高兴见到你！ # 有什么我可以帮你的吗？无论是回答问题、创作内容、查阅资料，还是随便聊 聊，我都在这里哦～ # 核心钩子方法 # 1. before_model（模型调用前） def before_model(self, state, runtime): """ 在模型调用前执行 返回值： - None: 继续正常流程 - dict: 更新状态（如 {"messages": [...]}） - {"jump_to": "..."}: 跳过正常流程 """ messages = state.get("messages", []) print(f"当前消息数: {len(messages)}") return None # 用途： # 消息修剪（trim messages） # PII 脱敏 # 输入验证 # 条件路由 # 1. before_model: 模型调用前的“守门员” # 执行时机： # 在 Agent 收集完所有上下文信息，准备将请求发送给大语言模型（LLM）之前立即执行。 # 核心参数： # state: 当前累积的对话状态（包含历史消息 messages、变量、用户输入等）。 # runtime: 运行时环境（包含配置、API Key、执行元数据等）。 # 主要功能场景： # 1. 动态提示词注入 (Prompt Injection)： # 根据当前状态，动态在 messages 列表头部插入 System Prompt 或 Few-shot 示例。 # 例：检测到用户情绪愤怒时，插入“请保持冷静和同理心”的系统指令。 # 2.上下文裁剪/压缩 (Context Pruning)： # 如果 state['messages'] 超过了模型的 Token 限制，在此处截断旧消息或调用摘要模型进行压缩。 # 3. 安全拦截 (Safety Guardrails)： # 检查用户输入是否包含敏感词、攻击性内容或越狱指令。如果违规，可以直接返回跳转指令，阻止请求发送给 LLM。 # 4.路由决策 (Routing)： # 分析用户意图，决定是否需要跳过当前模型，直接跳转到另一个特定的节点（Node）或工作流。 # 5.工具预填充： # 在某些架构中，可以预先计算某些工具的结果并放入上下文，让模型“看到”这些结果。 # 2. after_model（模型响应后） def after_model(self, state, runtime): """ 在模型响应后执行 返回值： - None: 不修改状态 - dict: 更新状态 """ # 统计调用次数 count = state.get("call_count", 0) return {"call_count": count + 1} # 用途： # 输出验证 # 格式化响应 # 统计信息 # 状态更新 # 2. after_model: 模型调用后的“质检员” # 执行时机： # 在 LLM 返回生成结果（Completion），但尚未将该结果正式存入状态或返回给用户之后立即执行。 # 核心参数： # state: 包含模型刚刚生成的原始响应（通常在 messages 的最后一条，或者在单独的 response 字段中）。 # runtime: 运行时环境。 # (部分框架还会传入 response 对象，包含 token 用量、延迟等元数据) # 主要功能场景： # 1.输出格式化与清洗： # 移除模型生成的多余标记（如 Thought:, Action: 等中间思维链，如果不需要展示给用户）。 # 修复 JSON 格式错误（如果模型本应输出 JSON 但漏了括号）。 # 2.幻觉检测与事实核查 (Hallucination Check)： # 调用外部知识库或搜索工具，验证模型回答的事实准确性。如果不准确，可以标记为需要重试。 # 3.合规性审查 (Compliance)： # 检查模型输出是否包含偏见、泄露隐私或违反公司政策的内容。 # 4.结构化提取 (Structured Extraction)： # 如果模型输出了非结构化的文本，在此处将其解析为结构化数据（如提取订单号、日期），并存入 state 供后续节点使用。 # 5.自动重试逻辑 (Self-Correction)： # 如果检测到模型输出不符合预期（例如没有调用应有的工具），可以修改 state 并指示框架重新调用模型（通过返回特定的状态标志或抛出重试信号）。 # 返回值的作用 # 返回 None # def before_model(self, state, runtime): # print("日志记录") # return None # 不做任何修改，继续流程 # 返回字典（更新状态） # def after_model(self, state, runtime): # count = state.get("count", 0) # return {"count": count + 1} # 更新状态中的 count # 返回 jump_to（控制流程） # def before_model(self, state, runtime): # if state.get("count", 0) > 10: # return {"jump_to": "__end__"} # 跳过模型，直接结束 # return None # jump_to 目标： # "__end__" - 结束 Agent # "tools" - 跳到工具节点 # 其他自定义节点 # 执行顺序（重要！） # agent = create_agent( # model=model, # middleware=[Middleware1(), Middleware2(), Middleware3()] # ) # 执行流程： # 1. Middleware1.before_model ↓ 正序 # 2. Middleware2.before_model ↓ # 3. Middleware3.before_model ↓ # [模型调用] # 6. Middleware3.after_model ↑ 逆序 # 5. Middleware2.after_model ↑ # 4. Middleware1.after_model ↑ # 类似洋葱模型：外层先进后出 # 用户请求 # ↓ # [ Middleware1.before ] <-- 外层入口 # ↓ # [ Middleware2.before ] # ↓ # [ Middleware3.before ] <-- 内层入口 # ↓ # [ 🤖 模型调用 (Model) ] # ↓ # [ Middleware3.after ] <-- 内层出口 (最先处理模型结果) # ↑ # [ Middleware2.after ] # ↑ # [ Middleware1.after ] <-- 外层出口 (最后处理结果) # ↑ # 返回给用户 # 实际应用 class TestMiddleware: # before_model 示例：添加情感安抚指令 def before_model(self, state, runtime): last_msg = state["messages"][-1] # 验证用户身份：如果用户不是授权用户，直接返回跳转指令 if last_msg["role"] != "user": return {"jump_to": "unauthorized_node"} # 安全检查：如果包含攻击性词汇，直接返回跳转指令 if "混蛋" in last_msg["content"]: return {"jump_to": "safety_block_node"} # 2. 动态注入：如果检测到用户很着急，注入安抚指令 if "着急" in last_msg["content"]: system_prompt = { "role": "system", "content": "【动态指令】用户非常着急，请优先安抚情绪并快速给出方案。", } # 插入到 system prompt 之后，历史记录之前 new_messages = [state["messages"][0], system_prompt] + state["messages"][1:] return {"messages": new_messages} return None # 正常流程 # after_model 示例：确保输出包含工单号 def after_model(self, state, runtime): # 获取模型最后生成的消息 response_content = state["messages"][-1].content if "投诉" in state.get("intent", ""): if "工单号" not in response_content: response_content += " 请包含工单号，方便我们定位问题。" return None # 接受输出 # 1. 日志中间件 class LogMiddleware: def before_model(self, state, runtime): print("日志记录：开始模型调用") return None # 不做任何修改，继续流程 def after_model(self, state, runtime): print("日志记录：模型调用结束") return None # 不做任何修改，继续流程 # 2. 计数中间件 class CallCounterMiddleware(AgentMiddleware): def after_model(self, state, runtime): count = state.get("model_call_count", 0) return {"model_call_count": count + 1} # # 需要 checkpointer 来保存自定义状态 # agent = create_agent( # model=model, # middleware=[CallCounterMiddleware()], # checkpointer=InMemorySaver() # ) # 3. 消息修剪中间件 class MessageTrimmerMiddleware(AgentMiddleware): def __init__(self, max_messages=5): super().__init__() self.max_messages = max_messages def before_model(self, state, runtime): messages = state.get("messages", []) if len(messages) > self.max_messages: # 只保留最近的 N 条消息 return {"messages": messages[-self.max_messages :]} return None # 4. 输出验证中间件 class OutputValidatorMiddleware(AgentMiddleware): def after_model(self, state, runtime): messages = state["messages"] last_msg = messages[-1] if last_msg["role"] == "assistant": # 验证模型输出是否符合要求 if "工单号" not in last_msg["content"]: # 触发跳转指令 return {"jump_to": "missing_order_id_node"} return None # 5. 限流中间件 class MaxCallsMiddleware(AgentMiddleware): def __init__(self, max_calls=10): super().__init__() self.max_calls = max_calls def before_model(self, state, runtime): count = state.get("model_call_count", 0) if count >= self.max_calls: # 触发跳转指令 return {"jump_to": "rate_limit_node"} return None def after_model(self, state, runtime): count = state.get("call_count", 0) return {"call_count": count + 1} # 内置中间件 # SummarizationMiddleware（自动摘要） from langchain.agents.middleware import SummarizationMiddleware def demo1(): agent = create_agent( model=chat_model, middleware=[ SummarizationMiddleware( model="groq:llama-3.1-8b-instant", # 可用便宜模型 max_tokens_before_summary=500, ) ], checkpointer=InMemorySaver(), ) # 作用： # 消息超过 token 限制时自动摘要 # 保留最近消息 + 旧消息摘要 # 详见 08_context_management 章节 # HumanInTheLoopMiddleware（人工审核） from langchain.agents.middleware import HumanInTheLoopMiddleware def demo2(): agent = create_agent( model=chat_model, middleware=[ HumanInTheLoopMiddleware( interrupt_on={"send_email": True} # 调用此工具前暂停 ) ], checkpointer=InMemorySaver(), ) # PIIMiddleware（敏感信息处理） from langchain.agents.middleware import PIIMiddleware def demo3(): agent = create_agent( model=chat_model, middleware=[ PIIMiddleware("email", strategy="redact"), # 邮箱脱敏 PIIMiddleware("phone_number", strategy="block"), # 电话拦截 # 输出阶段：防止 LLM 幻觉生成假的或泄露真的身份证号 PIIMiddleware("id_number", strategy="redact", direction="output"), ], ) # graph LR # User[用户输入] --> PII_Check{PIIMiddleware 检查} # PII_Check -- 发现 Email --> Redact[执行脱敏 redact] # Redact --> LLM[发送给大模型] # PII_Check -- 发现 手机号 --> Block[执行拦截 block] # Block --> Error[返回错误/拒绝服务] # PII_Check -- 无敏感信息 --> LLM # 常见问题 # 不能直接访问。before_model 和 after_model 只在模型节点执行。 # 如果需要拦截工具调用，使用 wrap_tool_call（高级特性）。 # 2. 多个中间件的顺序重要吗？ # 非常重要！ # middleware=[ # TrimmerMiddleware(), # 1. 先修剪消息 # SummarizationMiddleware(), # 2. 再摘要 # LoggingMiddleware() # 3. 最后记录日志 # ] # before_model 按列表顺序执行 # after_model 按列表逆序执行 # 3. 修改状态需要 checkpointer 吗？ # 自定义状态需要，messages 不需要： # # 不需要 checkpointer（messages 自动保存） # def after_model(self, state, runtime): # return {"messages": [...]} # # 需要 checkpointer（自定义字段） # def after_model(self, state, runtime): # return {"my_custom_field": 123} # 4. 能在中间件里调用另一个模型吗 # 可以，但要小心： # class ValidationMiddleware(AgentMiddleware): # def __init__(self): # self.validator_model = init_chat_model(...) # def after_model(self, state, runtime): # # 用另一个模型验证输出 # last_msg = state['messages'][-1] # validation_result = self.validator_model.invoke(...) # return None # 最佳实践 # # 1. 生产环境推荐配置 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # MessageTrimmerMiddleware(max_messages=20), # 限制消息数 # SummarizationMiddleware(model=..., max_tokens=2000), # 自动摘要 # LoggingMiddleware(), # 日志记录 # ], # checkpointer=SqliteSaver.from_conn_string("...") # ) # # 2. 开发环境 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # LoggingMiddleware(), # 只要日志 # ] # ) # # 3. 测试环境 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # MaxCallsMiddleware(max_calls=5), # 防止测试费用爆炸 # ] # ) if __name__ == "__main__": demo1()
""" LangChain 1.0 - Checkpointing (检查点持久化) ========================================= 本模块重点讲解： 1. SqliteSaver - SQLite 持久化（LangGraph 提供） 2. 与 InMemorySaver 的区别 3. 跨进程、跨重启的对话持久化 4. 实际应用场景 1. invoke 前：LangGraph 自动调用 checkpointer.get(thread_id="user_123") - 查询数据库，读取该 thread_id 的历史消息 - 如果是第一次，返回空列表 2. invoke 中：Agent 处理时会看到完整历史(默认确实会全部读取，) state = { "messages": [历史消息1, 历史消息2, 新消息] # 自动合并 } 3. invoke 后：LangGraph 自动调用 checkpointer.put(thread_id, state) - 将新的完整状态写入数据库 - 数据库存储：(thread_id, timestamp, messages) """ import sys import os from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver chat_model = get_chat_model() # 持久化数据库路径 db_path = "checkpoints.sqlite" # ============================================================================ # SqliteSaver 持久化 # ============================================================================ def example_sqlite_saver(): """ 示例2：使用 SqliteSaver 实现持久化 关键： 1. SqliteSaver.from_conn_string("sqlite:///path/to/db.sqlite") 2. 对话持久化到 SQLite 文件 3. 程序重启后仍可恢复 """ with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_zhangsan"}} # 第一次运行 response = agent.invoke( {"messages": [{"role": "user", "content": "你好,我叫张三"}]}, config ) print(response["messages"][-1].content) def example_sqlite_saver_2(): """ 示例3：使用 SqliteSaver 实现持久化 关键： 1. SqliteSaver.from_conn_string("sqlite:///path/to/db.sqlite") 2. 对话持久化到 SQLite 文件 3. 程序重启后仍可恢复 """ with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_zhangsan"}} # 第一次运行 reponse = agent.invoke( {"messages": [{"role": "user", "content": "你好,我是谁"}]}, config ) print(reponse["messages"][-1].content) # ============================================================================ # 带工具的持久化 Agent # ============================================================================ @tool def get_order_status(order_id: str) -> str: """查询订单状态""" orders = {"12345": "已发货，预计明天送达", "67890": "配送中，今天下午送达"} return orders.get(order_id, "订单不存在") """ 示例5：工具调用 + 持久化 Agent 记住工具调用历史 """ def example_3_tools_with_persistence(): """ 示例5：工具调用 + 持久化 Agent 记住工具调用历史 """ print("\n" + "=" * 70) print("示例 5：工具调用 + 持久化") print("=" * 70) db_path = "tools.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[get_order_status], checkpointer=checkpointer ) config = {"configurable": {"thread_id": "customer_001"}} print("\n第一轮：查询订单") print("客户: 查询订单 12345 的状态") response1 = agent.invoke( {"messages": [{"role": "user", "content": "查询订单 12345 的状态"}]}, config=config, ) print(f"Agent: {response1['messages'][-1].content}") print("\n第二轮：询问之前的查询结果") print("客户: 我的订单什么时候到？") response2 = agent.invoke( {"messages": [{"role": "user", "content": "我的订单什么时候到？"}]}, config=config, ) print(f"Agent: {response2['messages'][-1].content}") print("\n关键点：") print(" - Agent 记住了订单 12345 的查询结果") print(" - 工具调用历史也被持久化") print(" - 无需重复调用工具") # ============================================================================ # 实际应用 - 客服系统 # ============================================================================ def example_4_customer_service(): """ 示例6：实际应用 - 持久化客服系统 场景：客户可能分多次咨询，需要记住历史 """ print("\n" + "=" * 70) print("示例 6：实际应用 - 持久化客服系统") print("=" * 70) db_path = "customer_service.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[get_order_status], system_prompt="""你是客服助手。 特点： - 记住客户之前的咨询 - 友好、耐心 - 使用工具查询订单""", checkpointer=checkpointer, ) customer_id = "customer_zhang" config = {"configurable": {"thread_id": customer_id}} print("\n第一次咨询（今天上午）：") conversations_morning = ["你好，我想查询订单", "订单号是 12345"] for msg in conversations_morning: print(f"\n客户: {msg}") response = agent.invoke( {"messages": [{"role": "user", "content": msg}]}, config=config ) print(f"客服: {response['messages'][-1].content}") print("\n" + "-" * 70) print("[几个小时后...]") print("-" * 70) print("\n第二次咨询（今天下午）：") print("\n客户: 我的订单到哪了？") response = agent.invoke( {"messages": [{"role": "user", "content": "我的订单到哪了？"}]}, config=config, ) print(f"客服: {response['messages'][-1].content}") print("\n关键点：") print(" - 客户无需重复订单号") print(" - 系统记住了上午的咨询") print(" - 即使客服系统重启也不影响") print(" - 生产级应用的标准做法") # ============================================================================ # SqliteSaver 参数说明 # ============================================================================ # SqliteSaver 创建方式： # 1. from_conn_string + with 语句（推荐） # with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: # agent = create_agent(model=model, checkpointer=checkpointer) # agent.invoke(...) # - 自动管理连接和资源 # - 支持相对路径和绝对路径 # - 最简单安全的方式 # - 确保正确释放数据库连接 # - 注意：直接传文件路径，不要加 sqlite:/// 前缀 # 2. 使用 sqlite3.connect（高级） # import sqlite3 # conn = sqlite3.connect("checkpoints.sqlite") # checkpointer = SqliteSaver(conn) # - 需要手动管理连接 # - 适合需要自定义连接参数的场景 # 数据库文件路径： # - 相对路径：checkpoints.sqlite（当前目录） # - 绝对路径：C:/Users/xxx/data/checkpoints.sqlite（Windows） # - 内存数据库：:memory:（测试用，程序退出即丢失） # 最佳实践： # ✅ 始终使用 with 语句管理 SqliteSaver # ✅ 直接传文件路径，不要加 sqlite:/// 前缀 # ✅ 生产环境：使用绝对路径 # ✅ 开发测试：使用相对路径 # ✅ 单元测试：使用 :memory: # ✅ 定期备份数据库文件 # """ if __name__ == "__main__": example_sqlite_saver() example_sqlite_saver_2()
# 核心概念 # Checkpointing = 将对话状态持久化到数据库 # InMemorySaver → 内存中（程序退出即丢失） # SqliteSaver/ → SQLite 数据库（持久化存储） # 基本用法 # InMemorySaver 的限制 import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver chat_model = get_chat_model() def demo1(): agent = create_agent(model=chat_model, tools=[], checkpointer=InMemorySaver()) # 限制： # ❌ 程序重启后丢失 # ❌ 无法跨进程共享 # ❌ 不适合生产环境 # SqliteSaver（推荐生产使用） def SqliteSaver_demo(): db_path = "checkpoints.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_123"}} # 第一次运行 agent.invoke( {"messages": [{"role": "user", "content": "你好,我叫张三"}]}, config ) with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent(model=chat_model, checkpointer=checkpointer) result = agent.invoke( {"messages": [{"role": "user", "content": "我是谁"}]}, config ) print(result["messages"][-1].content) # 生成的checkpoints.sqlit文件 # ├── 09 - Checkpointing (检查点持久化)/ # │ └── check_point.py # ├── 📁 其他练习/ # ├── 📁 tools/ # ├── ⚙️ .env # ├── ️ checkpoints.sqlite # ├── 🗄️ checkpoints.sqlite-shm # ├── 🗄️ checkpoints.sqlite-wal # └── 📄 init_model.py # 运行结果 # 你是**张三**呀！👋 # 你之前已经跟我介绍过很多次啦（比如最开始你就说过“你好，我叫张三”）。 虽然你可能是在考我的记忆力，但我确实把你的名字记在心里了哦！😄 # 除了名字之外，我还挺好奇你的其他故事的。今天有什么特别想聊的话题，或 者需要我帮忙的地方吗？ # 跨进程访问 # 进程 A（Web 服务器） def demo2(): with SqliteSaver.from_conn_string("shared.sqlite") as checkpointer: agent_a = create_agent(model=model, checkpointer=checkpointer) agent_a.invoke({...}, config={"configurable": {"thread_id": "user_1"}}) # 进程 B（后台任务） with SqliteSaver.from_conn_string("shared.sqlite") as checkpointer: agent_b = create_agent(model=model, checkpointer=checkpointer) # 可以访问进程 A 创建的对话 agent_b.invoke({...}, config={"configurable": {"thread_id": "user_1"}}) # 参数说明 # SqliteSaver.from_conn_string() # 参数 说明 示例 # conn_string 数据库文件路径（不要加 sqlite:/// 前缀） "checkpoints.sqlite" # 路径格式 # 相对路径（当前目录） - 推荐 def demo3(): with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: agent = create_agent(model=model, checkpointer=checkpointer) # 绝对路径（生产环境） with SqliteSaver.from_conn_string("C:/data/checkpoints.sqlite") as checkpointer: agent = create_agent(model=model, checkpointer=checkpointer) # 重要： # ✅ 直接传文件路径，不要加 sqlite:/// 前缀 # ✅ 相对路径会在当前目录创建数据库 # ✅ Windows 路径使用正斜杠 / 或双反斜杠 \\ # 对比 InMemorySaver # 特性 InMemorySaver SqliteSaver # 持久化 ❌ 程序退出即丢失 ✅ 持久化到文件 # 跨进程 ❌ 无法共享 ✅ 可以共享 # 性能 ⚡ 快（内存） 🐢 慢一点（磁盘 I/O） # 适用 开发、测试 生产环境 # 实际应用 # 1.客服系统 def demo4(): # 客户今天上午咨询 with SqliteSaver.from_conn_string("customer_service.sqlite") as checkpointer: agent = create_agent(model=model, tools=[查询订单], checkpointer=checkpointer) config = {"configurable": {"thread_id": "customer_zhang"}} agent.invoke( {"messages": [{"role": "user", "content": "订单 12345 在哪？"}]}, config ) # 下午客户再次咨询（即使服务重启） with SqliteSaver.from_conn_string("customer_service.sqlite") as checkpointer: agent = create_agent(model=model, tools=[查询订单], checkpointer=checkpointer) agent.invoke({"messages": [{"role": "user", "content": "到了吗？"}]}, config) # Agent 记得上午查询的订单号！ # 多用户聊天 def demo5(): with SqliteSaver.from_conn_string("chat.sqlite") as checkpointer: agent = create_agent(model=model, checkpointer=checkpointer) # 用户A agent.invoke({...}, config={"configurable": {"thread_id": "user_alice"}}) # 用户B agent.invoke({...}, config={"configurable": {"thread_id": "user_bob"}}) # 所有用户的对话都持久化在 chat.sqlite 中 # 常见问题 # 1. 数据库文件在哪？ # # 相对路径 → 当前工作目录 # SqliteSaver.from_conn_string("sqlite:///checkpoints.sqlite") # # 文件位置：当前目录/checkpoints.sqlite # # 绝对路径 → 指定位置 # SqliteSaver.from_conn_string("sqlite:///C:/data/checkpoints.sqlite") # # 文件位置：C:/data/checkpoints.sqlite # 2. 如何清空某个用户的历史？ # 目前需要手动操作数据库： # import sqlite3 # conn = sqlite3.connect("checkpoints.sqlite") # cursor = conn.cursor() # # 删除特定 thread_id 的记录 # cursor.execute("DELETE FROM checkpoints WHERE thread_id = ?", ("user_123",)) # conn.commit() # conn.close() # 3. 数据库会无限增长吗？ # 会！需要定期清理： # 策略： # 定期删除旧对话（如 30 天前） # 限制每个 thread 的 checkpoint 数量 # 定期备份和归档 # 4. 性能影响？ # SQLite 比内存慢，但影响不大 # 适合中小型应用（< 10000 并发用户） # 大规模应用考虑 PostgreSQL（LangGraph 也支持） # 最佳实践 # 1. 生产环境使用绝对路径 + with 语句 # with SqliteSaver.from_conn_string("C:/production/data/checkpoints.sqlite") as checkpointer: # agent = create_agent(model=model, checkpointer=checkpointer) # # 2. 开发环境使用相对路径 # with SqliteSaver.from_conn_string("dev_checkpoints.sqlite") as checkpointer: # agent = create_agent(model=model, checkpointer=checkpointer) # # 3. 测试环境使用内存数据库 # with SqliteSaver.from_conn_string(":memory:") as checkpointer: # agent = create_agent(model=model, checkpointer=checkpointer) # # 4. 定期备份数据库文件 # # 使用系统任务定期复制 checkpoints.sqlite # # 5. 监控数据库大小 # import os # db_size = os.path.getsize("checkpoints.sqlite") # print(f"数据库大小: {db_size / 1024 / 1024:.2f} MB") # 核心要点 # InMemorySaver：内存存储，程序退出即丢失 # SqliteSaver：持久化到 SQLite 文件 # 创建方式：with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: # 路径格式：直接传文件路径，不要加 sqlite:/// 前缀 # 跨进程：多个进程可访问同一数据库 # 生产推荐：使用 SqliteSaver + with 语句 if __name__ == "__main__": SqliteSaver_demo()
""" LangChain 1.0 - Context Management (上下文管理) ============================================== 本模块重点讲解： 1. SummarizationMiddleware - 自动摘要中间件（LangChain 1.0 新增） 2. trim_messages - 消息修剪工具 3. 管理对话长度，避免超 token 4. 中间件的使用 """ from langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain_core.messages import trim_messages # 模拟一个长对话历史 from langchain_core.messages import HumanMessage, AIMessage import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model chat_model = get_chat_model() @tool def calculator(operation: str, a: float, b: float) -> str: """执行数学计算""" ops = { "add": lambda x, y: x + y, "multiply": lambda x, y: x * y, } result = ops.get(operation, lambda x, y: 0)(a, b) return f"{a} {operation} {b} = {result}" # 问题演示 - 对话历史无限增长 # ============================================================================ # 示例 1：解决方案 1 - SummarizationMiddleware（推荐） # ============================================================================ def create_agent_with_summarization(): """ 示例2：使用 SummarizationMiddleware 自动摘要 关键：LangChain 1.0 新增的中间件 当消息数超过阈值时，自动摘要旧消息 """ agent = create_agent( model=chat_model, tools=[calculator], checkpointer=InMemorySaver(), middleware=[ SummarizationMiddleware(model=chat_model, max_messages=2, trigger_tokens=30) ], # <--- 关键：设为一个小值，强制早期触发)], ) config = {"configurable": {"thread_id": "with_summary2"}} conversations = [ "计算 2 + 3", "结果再加10", "结果再加15", "结果再加20", "结果再加20", "结果再加20", "结果再加20", "结果再加20", "结果再加20", "请总结一下计算过程", ] for msg in conversations: print(f"\n用户: {msg}") response = agent.invoke( {"messages": [{"role": "user", "content": msg}]}, config=config ) print(response["messages"][-1].content) # 结果： # 用户: 计算 2 + 3 # 2 + 3 = 5 # 用户: 结果再加10 # 5 + 10 = 15 # 用户: 结果再加15 # 15 + 15 = 30 # 用户: 结果再加20 # 30 + 20 = 50 # 用户: 结果再加20 # 50 + 20 = 70 # 用户: 结果再加20 # 70 + 20 = 90 # 用户: 结果再加20 # 90 + 20 = 110 # 用户: 结果再加20 # 110 + 20 = 130 # 用户: 结果再加20 # 130 + 20 = 150 # 用户: 请总结一下计算过程 # ## 计算过程总结 # ### 初始计算 # - **第一步**：2 + 3 = **5** # ### 后续累加 # | 步骤 | 运算 | 结果 | # |------|------|------| # | 第二步 | 5 + 10 | 15 | # | 第三步 | 15 + 15 | 30 | # | 第四步 | 30 + 20 | 50 | # | 第五步 | 50 + 20 | 70 | # | 第六步 | 70 + 20 | 90 | # | 第七步 | 90 + 20 | 110 | # | 第八步 | 110 + 20 | 130 | # | 第九步 | 130 + 20 | **150** | # ### 总计 # - **初始值**：2 + 3 = 5 # - **累计增加**：10 + 15 + 20×6 = 145 # - **最终结果**：**150** # 共进行了 **9 次加法运算**，从最初的 5 逐步增加到最终的 150。 # 太棒了！🎉 SummarizationMiddleware 已经完美生效了！ # ✅ 为什么说它生效了？ # 你提供的输出结果是最好的证明： # 1.上下文长度限制：你设置了 max_messages=2。这意味着在最后一轮对话时，Agent 的“短期记忆”里实际上只保留了最后两条原始消息（即第9轮的“130+20”和第10轮的提问）。 # 2.信息完整性：如果没有摘要机制，Agent 应该完全忘记第1轮的 2+3=5 以及中间的累加过程，只能回答“我只记得最近加了20”。 # 3.最终表现：Agent 却能够精确地列出从第一步到第九步的所有细节，甚至算出了累计增加量（145）。 # 这说明：之前的那些被“移除”出短期记忆的对话，已经被中间件自动压缩成了一条高密度的摘要（Summary），并作为系统提示词传给了模型。 # 模型正是读取了这条摘要，才“回忆”起了整个计算链条。 # ============================================================================ # 示例 2：理解 SummarizationMiddleware 参数 # ============================================================================ # def demo_summarization_middleware(): # SummarizationMiddleware( # model: BaseChatModel, # max_messages: int = 20, # trigger_tokens: int | None = None, # summary_prompt: ChatPromptTemplate | None = None, # ) # model BaseChatModel 必填 (无默认) 指定执行摘要任务的模型实例。 • 省钱策略：主 Agent 用 gpt-4o，摘要用 gpt-4o-mini。 # • 一致性：通常直接使用 Agent 的主模型。 # max_messages int 20 (视版本而定) 触发总结时，保留最近多少条原始消息不压缩。 • 强逻辑/数学：设为 2 ~ `4`。保留最近几步推理，防止逻辑断层。 • <v>**<v>创作/聊天<v>**<v>：设为 `10` ~ 20。保留更多近期上下文以保持连贯性。 # trigger_tokens int / None None (通常为窗口 75%) 上下文 Token 数超过此值时，强制触发摘要。 • 调试/测试：设为 100 ~ `200`。强制在短对话中触发以便观察效果。 • <v>**<v>生产环境<v>**<v>：设为模型上限的 `50%` ~ 70%。平衡性能与成本。 # summary_prompt ChatPromptTemplate 内置默认模板 自定义如何生成摘要的指令模板。 • 特定领域：若需提取 JSON、代码或特定实体，需自定义此 Prompt。 # • 默认行为：通常只需总结关键事实、待办和实体，忽略闲聊。 # ============================================================================ # 示例 3：手动消息修剪（trim_messages） # ============================================================================ from langchain_core.messages import ( HumanMessage, AIMessage, SystemMessage, trim_messages, ) def demo1_trim_messages(): print("=== 开始演示上下文裁剪 ===") # 1. 初始化 Agent (保持你的配置) # 注意：如果你手动 trim 了，Middleware 可能不会触发，这是正常的。 agent = create_agent( model=chat_model, checkpointer=InMemorySaver(), middleware=[ SummarizationMiddleware(model=chat_model, max_messages=2, trigger_tokens=30) ], ) config = {"configurable": {"thread_id": "test_trim_final"}} # 2. 构造长历史消息 messages = [ HumanMessage(content="消息 1：今天天气不错。"), AIMessage(content="回复 1：是的，适合出门。"), HumanMessage(content="消息 2：我想去公园。"), AIMessage(content="回复 2：公园是个好主意。"), HumanMessage(content="消息 3：但是公园有点远。"), AIMessage(content="回复 3：那我们可以去近一点的地方。"), HumanMessage(content="消息 4：附近有什么推荐吗？"), AIMessage(content="回复 4：附近的植物园很不错。"), ] print(f"原始消息数量: {len(messages)}") # 3. 执行裁剪 # 目标：保留最后的部分，但限制 Token 数量 trimmed = trim_messages( messages, max_tokens=30, # 设一个小值，强制它丢弃前面的“消息 1/2” token_counter="approximate", strategy="last", start_on="human", # 确保第一条是用户说的 ) print(f"裁剪后消息数量: {len(trimmed)}") print("--- 裁剪后的上下文内容 ---") for i, msg in enumerate(trimmed): role = "User" if isinstance(msg, HumanMessage) else "AI" print(f"[{i}] {role}: {msg.content}") print("--------------------------") # 4. 【关键】构建输入 # 必须把 trimmed 列表作为一个整体历史传进去，而不是循环单条发送 # 我们再加一个新的问题，测试 Agent 是否记得前面的内容 new_question = HumanMessage(content="结合刚才的对话，你觉得去植物园好还是公园好？") # 合并历史和新问题 input_payload = {"messages": trimmed + [new_question]} print("\n正在调用 Agent (带着裁剪后的历史 + 新问题)...") try: response = agent.invoke(input_payload, config=config) # 获取最后一条消息（AI 的回答） final_answer = response["messages"][-1] print("\n=== Agent 的最终回答 ===") print(final_answer.content) print("========================") # 验证：如果回答里提到了“植物园”或“公园”，说明上下文保留成功了 if "植物园" in final_answer.content or "公园" in final_answer.content: print("✅ 成功：Agent 记住了裁剪后的上下文！") else: print("⚠️ 警告：Agent 的回答似乎没有包含上下文信息。") except Exception as e: print(f"❌ 调用失败: {e}") import traceback traceback.print_exc() ##结果 # === 开始演示上下文裁剪 === # 原始消息数量: 8 # 裁剪后消息数量: 2 # --- 裁剪后的上下文内容 --- # [0] User: 消息 4：附近有什么推荐吗？ # [1] AI: 回复 4：附近的植物园很不错。 # -------------------------- # 正在调用 Agent (带着裁剪后的历史 + 新问题)... # === Agent 的最终回答 === # 结合刚才的对话，既然我之前特意推荐了植物园，那我依然觉得**植物园**会更好一些。 # 相比普通公园，植物园通常有以下优势： # 1. **景观更丰富**：植物种类更多，四季景色变化更明显，观赏性强。 # 2. **环境更清幽**：一般比普通公园更适合静心散步、呼吸新鲜空气。 # 3. **体验更独特**：之前提到它“很不错”，说明它有独特的亮点，不仅仅是普 通的绿地。 # 当然，如果你只是想进行简单的运动或带孩子玩耍，普通公园也很方便；但如果是为了放松身心、欣赏自然，植物园会是更优选。 # ======================== # ✅ 成功：Agent 记住了裁剪后的上下文！ # ============================================================================ # 示例 4：对比不同策略 # ============================================================================ def example_4_comparison(): """ 策略对比： 1. 不做处理（默认） 优点：保留完整历史 缺点：会超 token、成本高 适用：短对话 2. SummarizationMiddleware（推荐） 优点： - 自动化，无需手动管理 - 保留重要信息（通过摘要） - 平滑过渡 缺点： - 摘要可能丢失细节 - 额外的摘要成本 适用：长对话、需要保留上下文 3. trim_messages（手动修剪） 优点： - 精确控制 - 简单直接 - 无额外成本 缺点： - 旧消息完全丢失 - 可能断开上下文 适用：只需要最近 N 轮 4. 滑动窗口（自定义） 优点： - 保留系统消息 + 最近消息 - 可控成本 缺点： - 需要自己实现 适用：有明确规则的场景 推荐方案： - 短对话（<10轮）：不处理 - 中长对话：SummarizationMiddleware - 只要最近几轮：trim_messages """ # ============================================================================ # 示例 5：实际应用 - 客服机器人 # ============================================================================ def example_5_customer_service(): """ 客服机器人场景： 1. 问题：“我想知道北京天气” 2. 客服：“好的，我会查询北京的天气。” 3. 问题：“我想知道北京的温度” 4. 客服：“北京当前天气晴朗，温度20°C，空气质量良好。” 5. 问题：“我想知道北京的湿度” 6. 客服：“我没有查询到北京的湿度信息。” 场景：客服对话可能很长，需要管理上下文 """ agent = create_agent( model=chat_model, tools=[calculator], system_prompt="""你是客服助手。 1. 你是一个客服助手，负责回答用户的问题。 2. 你只能使用提供的工具（如计算器）来回答问题。 3. 你不能编造信息，只能基于提供的上下文回答。 4. 你不能回答与客服助手无关的问题。 """, checkpointer=InMemorySaver(), middleware=[ SummarizationMiddleware( model=chat_model, trim_tokens_to_summarize=800, ) ], ) config = {"configurable": {"thread_id": "customer_123"}} # 模拟客服对话 conversations = [ "你好，我想咨询订单", "我的订单号是 12345", "帮我算一下 100 乘以 2 的优惠价", "谢谢", ] for msg in conversations: print(f"\n客户: {msg}") response = agent.invoke( {"messages": [{"role": "user", "content": msg}]}, config=config ) print(f"客服: {response['messages'][-1].content}") print(f"\n总消息数: {len(response['messages'])}") print("\n关键点：") print(" - 自动管理对话长度") print(" - 重要信息（订单号）通过摘要保留") print(" - 适合生产环境") if __name__ == "__main__": # create_agent_with_summarization() # demo1_trim_messages() example_5_customer_service()
""" LangChain 1.0 - Memory Basics (内存管理基础) ========================================== 本模块重点讲解 LangGraph 提供的内存管理机制，包括： 1. InMemorySaver - LangGraph 提供的内存管理 2. checkpointer 参数 - 为 Agent 添加内存 3. thread_id - 会话管理 4. 多轮对话状态保持 """ from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model from langchain_core.tools import tool from init_model import get_chat_model from langchain.agents import create_agent # 导入自定义工具 from tools import weather from tools import web_search from tools import calculator # 初始化数据 weather_tool = weather.get_weather web_search_tool = web_search.web_search calculator_tool = calculator.calculator chat_model = get_chat_model() # 创建一个简单的工具 @tool def get_user_info(user_id: str) -> str: """获取用户信息""" users = {"123": "张三，25岁，工程师", "456": "李四，30岁，设计师"} return users.get(user_id, "用户不存在") # ============================================================================ # 示例 简单客服机器人 # ============================================================================ def demo_simple_customer_service(): """ 演示简单客服机器人 """ agent = create_agent( model=chat_model, tools=[get_user_info], checkpointer=InMemorySaver(), # 添加内存 ) config = {"configurable": {"thread_id": "session_1"}} print("\n第一轮：查询用户信息") response1 = agent.invoke( {"messages": [{"role": "user", "content": "查询用户 123 的信息"}]}, config=config, ) print(f"Agent: {response1['messages'][-1].content}") print("\n第二轮：询问之前的信息") response2 = agent.invoke( {"messages": [{"role": "user", "content": "你之前查询的用户 123 信息是？"}]}, config=config, ) print(f"Agent: {response2['messages'][-1].content}") if __name__ == "__main__": demo_simple_customer_service()
# 核心概念 # 内存 = Agent 记住对话历史的能力 # 默认情况下，每次调用 agent.invoke() 都是全新的开始，不记得之前的对话。使用 InMemorySaver 可以让 Agent 记住历史。 # 基本用法 # 没有内存（默认） from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model from langchain_core.tools import tool from init_model import get_chat_model from langchain.agents import create_agent # 导入自定义工具 from tools import weather from tools import web_search from tools import calculator # 初始化数据 weather_tool = weather.get_weather web_search_tool = web_search.web_search calculator_tool = calculator.calculator chat_model = get_chat_model() def demo_no_memory(): """ 演示没有内存的 Agent """ agent = create_agent( model=chat_model, ) # 第一轮 response = agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}) print(response) # 第二轮 - 不记得第一轮！ response = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}) # 添加内存 def demo_with_memory(): # 1. 创建 Agent 时添加 checkpointer agent = create_agent( model=chat_model, tools=[], checkpointer=InMemorySaver() # 添加内存 ) # 2. 调用时指定 thread_id config = {"configurable": {"thread_id": "conversation_1"}} # 第一轮 agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config=config) # 第二轮 - 记得第一轮！ response = agent.invoke( {"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config ) print(response["messages"][-1].content) # 关键参数 # checkpointer # 作用：为 Agent 添加内存管理能力 def demo1(): from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model=chat_model, # tools=tools, checkpointer=InMemorySaver(), # InMemorySaver = 短期内存 ) # 注意： # InMemorySaver - ��存中保存（进程结束就丢失） # 后续会学习持久化（SQLite、Postgres） # thread_id # 作用：区分不同的会话 # 每个 thread_id 对应一个独立的会话 （会话之间互不干扰） def demo2(): agent = create_agent( model=chat_model, # tools=tools, checkpointer=InMemorySaver(), # InMemorySaver = 短期内存 ) # 会话 1 config1 = {"configurable": {"thread_id": "user_alice"}} agent.invoke({...}, config=config1) # 会话 2（完全独立） config2 = {"configurable": {"thread_id": "user_bob"}} agent.invoke({...}, config=config2) # thread_id 的选择： # 聊天应用：使用用户 ID 或会话 ID # 多轮任务：使用任务 ID # 测试：使用描述性字符串（如 "test_1"） # 内存保存了什么？ # agent.invoke({"messages": [{"role": "user", "content": "你好"}]}, config) # InMemorySaver 保存： # { # "thread_id": "xxx", # "messages": [ # HumanMessage("你好"), # AIMessage("你好！有什么可以帮助你的吗？") # ] # } # agent.invoke({"messages": [{"role": "user", "content": "天气"}]}, config) # InMemorySaver 更新： # { # "thread_id": "xxx", # "messages": [ # HumanMessage("你好"), # AIMessage("你好！有什么可以帮助你的吗？"), # HumanMessage("天气"), # AIMessage("...") # ] # } # 自动追加历史 # 你只需要传新消息 # agent.invoke( # {"messages": [{"role": "user", "content": "新问题"}]}, # config # ) # checkpointer 自动： # 1. 读取之前的历史 # 2. 追加新消息 # 3. 调用模型（传入完整历史） # 4. 保存新的历史 # 多会话管理 # 场景：多用户聊天 def demo3(): agent = create_agent(model=chat_model, tools=[], checkpointer=InMemorySaver()) # 用户 Alice config_alice = {"configurable": {"thread_id": "user_alice"}} agent.invoke({"messages": [...]}, config_alice) # 用户 Bob config_bob = {"configurable": {"thread_id": "user_bob"}} agent.invoke({"messages": [...]}, config_bob) # 两个会话完全独立 # 场景：同一用户的不同任务 def demo4(): agent = create_agent(model=chat_model, tools=[], checkpointer=InMemorySaver()) # 用户 Alice config_alice = {"configurable": {"thread_id": "user_alice"}} agent.invoke({"messages": [...]}, config_alice) # 任务 1 config_task1 = {"configurable": {"thread_id": "task_1"}} agent.invoke({"messages": [...]}, config_task1) # 任务 2 config_task2 = {"configurable": {"thread_id": "task_2"}} agent.invoke({"messages": [...]}, config_task2) # 两个任务互不干扰 # 内存 + 工具 # Agent 会记住工具调用的结果： def demo5(): agent = create_agent( model=chat_model, tools=[weather_tool], checkpointer=InMemorySaver() ) config = {"configurable": {"thread_id": "session_1"}} # 第一轮：使用工具 agent.invoke({"messages": [{"role": "user", "content": "北京"}]}, config) # 第二轮：使用工具结果 response = agent.invoke( {"messages": [{"role": "user", "content": "刚刚查询的结果是什么"}]}, config ) for msg in response["messages"][-5:]: print(f"{msg.__class__.__name__}: {msg.content}") # AIMessage: # ToolMessage: 晴天，温度 20°C，空气质量良好 # AIMessage: 北京当前天气晴朗，温度20°C，空气质量良好。 # HumanMessage: 刚刚查询的结果是什么 # AIMessage: 刚刚查询的结果是：北京当前天气晴朗，温度20°C，空气质量良好。 # 常见问题 # 1. 为什么 Agent 不记得？ # 检查： # ✅ 是否添加了 checkpointer=InMemorySaver()？ # ✅ 是否传入了 config 参数？ # ✅ 两次调用的 thread_id 是否相同？ # # ❌ 错误：没有 checkpointer # agent = create_agent(model=model, tools=[]) # agent.invoke({...}) # 不会记住 # # ❌ 错误：没有 config # agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver()) # agent.invoke({...}) # 不会记住 # # ❌ 错误：thread_id 不同 # agent.invoke({...}, {"configurable": {"thread_id": "1"}}) # agent.invoke({...}, {"configurable": {"thread_id": "2"}}) # 不同会话 # # ✅ 正确 # agent = create_agent(model=model, tools=[], checkpointer=InMemorySaver()) # config = {"configurable": {"thread_id": "1"}} # agent.invoke({...}, config) # agent.invoke({...}, config) # 记得！ # 2. InMemorySaver 会丢失数据吗？ # 会！ InMemorySaver 只保存在内存中： # ✅ 同一进程内有效 # ❌ 程序重启后丢失 # ❌ 不同进程无法共享 # 解决方案：Module 09 会学习持久化（SQLite） # 3. 内存会无限增长吗？ # 会！ 默认情况下，InMemorySaver 会保存所有消息。 # 问题： # 消息越来越多 # 超过模型的 token 限制 # 响应变慢、成本增加 # 解决方案：Module 08 会学习上下文管理（修剪、摘要） # 4. 如何清空某个会话的历史？ # 目前 InMemorySaver 没有提供删除 API。 # 临时方案： # 使用新的 thread_id # 或重新创建 Agent # 实际应用场景 # 1. 聊天机器人 def chat_bot(user_id: str, message: str, agent): config = {"configurable": {"thread_id": f"user_{user_id}"}} response = agent.invoke( {"messages": [{"role": "user", "content": message}]}, config ) return response["messages"][-1].content # 2. 多轮任务助手 def process_task(task_id: str, user_input: str): config = {"configurable": {"thread_id": f"task_{task_id}"}} response = agent.invoke( {"messages": [{"role": "user", "content": user_input}]}, config ) return response["messages"][-1].content # 3. 客服系统 def customer_service(user_id: str, message: str): agent = create_agent( model=chat_model, tools=[order_tool], system_prompt="你是客服助手，记住用户的订单号", checkpointer=InMemorySaver(), ) config = {"configurable": {"thread_id": f"customer_{user_id}"}} response = agent.invoke( {"messages": [{"role": "user", "content": message}]}, config ) return response["messages"][-1].content # 核心要点 # 默认无内存：每次 invoke 是全新开始 # 添加内存：checkpointer=InMemorySaver() # 会话管理：config={"configurable": {"thread_id": "xxx"}} # 自动保存：checkpointer 自动管理历史 # 多会话：不同 thread_id = 不同会话 # 记住工具：也会记住工具调用结果 if __name__ == "__main__": # demo5() agent = create_agent( model=chat_model, tools=[weather_tool], checkpointer=InMemorySaver() ) # 聊天机器人 print(chat_bot("alice", "北京天气", agent)) print(chat_bot("alice", "刚刚的结果是什么", agent))
"""LangChain 1.0 - Agent 执行循环本模块重点讲解：Agent 执行循环的详细过程流式输出（streaming）查看中间步骤理解消息流转"""import sysimport osfrom urllib import responsefrom openai import chatsys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))from init_model import get_chat_modelfrom langchain_core.tools import toolfrom init_model import get_chat_modelfrom langchain.agents import create_agent导入自定义工具from tools import weatherfrom tools import web_searchfrom tools import calculator初始化数据weather_tool = weather.get_weatherweb_search_tool = web_search.web_searchcalculator_tool = calculator.calculatorchat_model = get_chat_model()============================================================================示例 1：理解执行循环 - 查看完整消息历史============================================================================def example_1():""" 示例1：理解执行循环 - 查看完整消息历史 展示 Agent 执行过程中的所有消息，包括工具调用和模型回复 """ agent = create_agent(model=chat_model, tools=[calculator_tool]) # 计算工具 response = agent.invoke({"messages": [{"role": "user", "content": "25 乘以 8"}]}) # 1. 查看完整历史 for msg in response["messages"]: print(f"{msg.__class__.__name__}: {msg.content}") # 2.查看结果 print(response["messages"][-1].content) ============================================================================示例 2：流式输出（Streaming）============================================================================def example_2():""" 示例2：流式输出（Streaming） 展示 Agent 执行过程中的实时输出，适用于长任务 """ agent = create_agent( model=chat_model, tools=[calculator_tool, weather_tool] ) # 计算工具 #天气查询工具 message = [ { "role": "user", "content": "北京今天天气怎么样？", } ] for chunk in agent.stream({"messages": message}): # chunk 结构可能是 {'model': {'messages': [...]}} 或 {'tools': {'messages': [...]}} # 我们需要遍历 chunk 的值来找到 'messages' 列表 for key, value in chunk.items(): if isinstance(value, dict) and "messages" in value: messages_list = value["messages"] for msg in messages_list: # 获取内容 content = getattr(msg, "content", "") # 1. 处理 AI 的回答 (流式或最终) if msg.type == "ai" and content: # 使用 end="" 实现打字机效果 print(content, end="", flush=True) # 2. 处理工具的执行结果 (可选，如果你想看中间计算过程) elif msg.type == "tool" and content: # 换行显示工具结果，避免和 AI 回答混在一起 print(f"\n[计算器结果]: {content}", flush=True) # 3. 处理工具调用意图 (可选，显示模型决定调用什么) elif ( msg.type == "ai" and hasattr(msg, "tool_calls") and msg.tool_calls ): tool_name = msg.tool_calls[0].get("name", "unknown") print(f"\n[模型决定调用工具]: {tool_name}...", flush=True) ============================================================================示例 3：多步骤执行============================================================================def example_3():""" 示例3：Agent 执行多个工具调用 理解复杂任务的执行过程 """ agent = create_agent( model=chat_model, tools=[calculator_tool], system_prompt="你是一个数学助手。当遇到复杂计算时，分步骤计算。", ) response = agent.invoke( {"messages": [{"role": "user", "content": "先算 10 加 20，然后把结果乘以 3"}]} ) tool_calls_count = 0 for msg in response["messages"]: if hasattr(msg, "tool_calls") and msg.tool_calls: tool_calls_count += len(msg.tool_calls) print(f"工具调用 {tool_calls_count}: {msg.tool_calls}") print(f"\n工具调用次数: {tool_calls_count}") print(f"最终答案: {response['messages'][-1].content}") ============================================================================示例 4：理解消息类型============================================================================def example_4():""" 示例4：理解消息类型 展示不同消息类型的作用和区别 """ agent = create_agent( model=chat_model, tools=[calculator_tool, weather_tool] ) # 计算工具 #天气查询工具 response = agent.invoke( {"messages": [{"role": "user", "content": "北京今天天气怎么样？"}]} ) print("\n消息类型分析：") for msg in response["messages"]: msg_type = msg.__class__.__name__ if msg_type == "HumanMessage": print(f"\n[HumanMessage] 用户输入") print(f" 内容: {msg.content}") elif msg_type == "AIMessage": if hasattr(msg, "tool_calls") and msg.tool_calls: print(f"\n[AIMessage] AI 决定调用工具") print(f" 工具: {msg.tool_calls[0]['name']}") print(f" 参数: {msg.tool_calls[0]['args']}") else: print(f"\n[AIMessage] AI 的最终回答") print(f" 内容: {msg.content}") elif msg_type == "ToolMessage": print(f"\n[ToolMessage] 工具执行结果") print(f" 工具: {msg.name}") print(f" 结果: {msg.content}") print("\n\n消息类型总结：") print( """ HumanMessage → 用户的输入 AIMessage → AI 的输出（可能包含 tool_calls 或最终答案） ToolMessage → 工具的执行结果 SystemMessage → 系统指令（通过 system_prompt 设置） """ ) if name == "__main__":# example_1() # example_2() example_4()
# 核心概念 # Agent 执行循环 = 自动化的"思考-行动-观察"过程 # Agent 不是一次性调用，而是一个循环： # 用户问题 → AI 思考 → 调用工具 → 观察结果 → 继续思考 → 最终答案 # 执行循环详解 # 执行循环详解 # ┌─────────────┐ # │ 用户提问 │ # │ HumanMessage│ # └──────┬──────┘ # ↓ # ┌─────────────┐ # │ AI 分析问题 │ # │ 需要工具？ │ # └──────┬──────┘ # ↓ 是 # ┌─────────────┐ # │ AI 决定调用 │ # │ AIMessage │ # │ (tool_calls)│ # └──────┬──────┘ # ↓ # ┌─────────────┐ # │ 执行工具 │ # │ ToolMessage │ # └──────┬──────┘ # ↓ # ┌─────────────┐ # │ AI 看结果 │ # │ 生成答案 │ # │ AIMessage │ # └─────────────┘ import sys import os from urllib import response from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model from langchain_core.tools import tool from init_model import get_chat_model from langchain.agents import create_agent # 导入自定义工具 from tools import weather from tools import web_search from tools import calculator # 初始化数据 weather_tool = weather.get_weather web_search_tool = web_search.web_search calculator_tool = calculator.calculator chat_model = get_chat_model() # 使用工具完整过程 def demo1(): agent = create_agent(model=chat_model, tools=[calculator_tool]) # 计算工具 response = agent.invoke({"messages": [{"role": "user", "content": "25 乘以 8"}]}) # 1. 查看完整历史 for msg in response["messages"]: print(f"{msg.__class__.__name__}: {msg.content}") # 2. 获取最终答案 final_answer = response["messages"][-1].content print(f"最终答案: {final_answer}") # 3. 查看使用的工具 used_tools = [] for msg in response["messages"]: if hasattr(msg, "tool_calls") and msg.tool_calls: for tc in msg.tool_calls: used_tools.append(tc["name"]) print(f"使用的工具: {used_tools}") # 流式输出（Streaming） # 用于实时显示 Agent 的进度 def demo2_fixed(): agent = create_agent(model=chat_model, tools=[calculator_tool]) print("\n--- 开始流式输出 ---") for chunk in agent.stream({"messages": [{"role": "user", "content": "25 乘以 8"}]}): # chunk 结构可能是 {'model': {'messages': [...]}} 或 {'tools': {'messages': [...]}} # 我们需要遍历 chunk 的值来找到 'messages' 列表 for key, value in chunk.items(): if isinstance(value, dict) and "messages" in value: messages_list = value["messages"] for msg in messages_list: # 获取内容 content = getattr(msg, "content", "") # 1. 处理 AI 的回答 (流式或最终) if msg.type == "ai" and content: # 使用 end="" 实现打字机效果 print(content, end="", flush=True) # 2. 处理工具的执行结果 (可选，如果你想看中间计算过程) elif msg.type == "tool" and content: # 换行显示工具结果，避免和 AI 回答混在一起 print(f"\n[计算器结果]: {content}", flush=True) # 3. 处理工具调用意图 (可选，显示模型决定调用什么) elif ( msg.type == "ai" and hasattr(msg, "tool_calls") and msg.tool_calls ): tool_name = msg.tool_calls[0].get("name", "unknown") print(f"\n[模型决定调用工具]: {tool_name}...", flush=True) print("\n--- 结束 ---") # --- 开始流式输出 --- # [模型决定调用工具]: calculator... # [计算器结果]: 25.0 multiply 8.0 = 200.0 # 25 乘以 8 等于 200。 # --- 结束 --- # stream vs invoke # 方法 返回 用途 # invoke() 完整结果 等待完成后一次性获取 # stream() 生成器 实时获取中间步骤 # 多步骤执行 # Agent 可以多次调用工具： def demo3(): agent = create_agent(model=chat_model, tools=[calculator_tool]) # 问题：先算 10 + 20，然后乘以 3 response = agent.invoke( {"messages": [{"role": "user", "content": "先算 10 + 20，然后乘以 3"}]} ) # Agent 可能会： # 1. 调用 calculator(add, 10, 20) → 30 # 2. 调用 calculator(multiply, 30, 3) → 90 # 3. 返回最终答案 # 查看完整历史 for i, msg in enumerate(response["messages"], 1): print(f"\n--- 消息 {i}: {msg.__class__.__name__} ---") if hasattr(msg, "content"): print(f"内容: {msg.content}") if hasattr(msg, "tool_calls") and msg.tool_calls: for tc in msg.tool_calls: print(f"工具: {tc['name']}, 参数: {tc['args']}") # 使用 stream 查看步骤 def demo4(chunk): agent = create_agent(model=chat_model, tools=[calculator_tool]) step = 0 input_data = {"messages": [{"role": "user", "content": "先算 10 + 20，然后乘以 3"}]} print("--- 开始执行多步计算 ---") for chunk in agent.stream(input_data): step += 1 print(f"步骤 {step}:") # 【关键修改】：messages 不在 chunk 顶层，而在 chunk['model'] 或 chunk['tools'] 里 messages_list = None for value in chunk.values(): if isinstance(value, dict) and "messages" in value: messages_list = value["messages"] break if messages_list: latest = messages_list[-1] print(f" 类型: {latest.__class__.__name__}") # 打印内容 if hasattr(latest, "content") and latest.content: print(f" 内容: {latest.content}") elif hasattr(latest, "content"): print(f" 内容: (空，可能是纯工具调用)") # 打印工具调用详情 if hasattr(latest, "tool_calls") and latest.tool_calls: for tc in latest.tool_calls: print(f" 🛠️ 工具: {tc['name']}, 参数: {tc['args']}") else: print(" (未解析到消息)") print("--- 结束 ---") # 查看是否调用了工具 def check_agent_tool_usage(): # 1. 创建 Agent (确保 chat_model 和 calculator_tool 已定义) agent = create_agent(model=chat_model, tools=[calculator_tool]) # 2. 执行任务 (invoke 会阻塞直到完成) response = agent.invoke( {"messages": [{"role": "user", "content": "先算 10 + 20，然后乘以 3"}]} ) # 3. 检查消息列表中是否有 tool_calls # response["messages"] 包含了整个对话历史 (用户问 -> AI思考/调工具 -> 工具返回 -> AI回答) has_tool_calls = any( hasattr(msg, "tool_calls") and msg.tool_calls for msg in response["messages"] ) if has_tool_calls: print("✅ Agent 使用了工具") # 进阶：打印具体用了什么工具 for msg in response["messages"]: if hasattr(msg, "tool_calls") and msg.tool_calls: for tc in msg.tool_calls: print(f" - 调用了: {tc['name']}, 参数: {tc['args']}") else: print("❌ Agent 直接回答 (未使用工具)") # . Agent 可以调用多少次工具？ # 答：默认没有限制，直到得到最终答案 # 但可能会： # 超时 # 达到 token 限制 # 模型决定停止 # 3. 如何限制工具调用次数？ # LangChain 1.0 的 create_agent 默认使用 LangGraph，可以通过配置限制： def create_agent_with_tool_limit(model, tools, max_tool_calls=3): """ 创建一个 Agent，限制工具调用次数 :param model: 语言模型 :param tools: 可用工具列表 :param max_tool_calls: 最大允许调用次数 :return: 配置好的 Agent """ return create_agent( model=model, tools=tools, # 配置 LangGraph 限制 langgraph_config={ "max_iterations": max_tool_calls, }, ) if __name__ == "__main__": check_agent_tool_usage()
目录结构 init_model.py 和 tools文件在历史文章中05 - Simple Agent (简单 Agent)├── tools │ ├── __pycache__ │ ├── __init__.py │ ├── calculator.py │ ├── weather.py │ └── web_search.py ├── main.py ├── 其他练习 ├── .env └── init_model.py 创建 Agent并调用工具""" LangChain 1.0 - Simple Agent (使用 create_agent) ============================================= 本模块重点讲解： 1. 使用 create_agent 创建 Agent（LangChain 1.0 统一API） 2. Agent 自动决定何时使用工具 3. Agent 执行循环的工作原理 """ import sys import os from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model from langchain_core.tools import tool from init_model import get_chat_model from langchain.agents import create_agent # 导入自定义工具 from tools import weather from tools import web_search from tools import calculator # 初始化数据 weather_tool = weather.get_weather web_search_tool = web_search.web_search calculator_tool = calculator.calculator chat_model = get_chat_model() def example_1_basic_agent(): """ 示例1：基本 Agent 演示 """ # 创建 Agent agent = create_agent(model=chat_model, tools=[weather_tool]) # 只给一个工具 # 测试：需要工具的问题 print("\n测试1：询问天气（需要工具）") response = agent.invoke( {"messages": [{"role": "user", "content": "北京今天天气怎么样？"}]} ) final_message = response["messages"][-1] print(final_message.content) # 测试 不需要工具的问题 print("\n测试2：简单介绍（不需要工具）") response = agent.invoke( { "messages": [ {"role": "user", "content": "你好，请简单介绍一下你自己，20字以内"} ] } ) final_message = response["messages"][-1] print(final_message.content) # ============================================================================ # 示例 2：多工具 Agent # ============================================================================ def example_2_multi_tools_agent(): """ 示例2：多工具 Agent 演示 """ # 创建 Agent agent = create_agent( model=chat_model, tools=[weather_tool, web_search_tool, calculator_tool] ) print("\n配置的工具：") print(" - get_weather（天气查询）") print(" - calculator（计算器）") print(" - web_search（网页搜索）") # 测试不同类型的问题 tests = [ "上海的天气怎么样？", # 应该用 get_weather "15 乘以 23 等于多少？", # 应该用 calculator ] for test in tests: print(f"\n测试：{test}") response = agent.invoke({"messages": [{"role": "user", "content": test}]}) final_message = response["messages"][-1] print(final_message.content) print("\n关键点：") print(" - Agent 从多个工具中选择最合适的") print(" - 基于工具的 docstring 理解工具用途") # ============================================================================ # 示例 3：带系统提示的 Agent # ============================================================================ def example_3_system_prompt_agent(): """ 示例3：带系统提示的 Agent 使用 system_prompt 参数 """ # 创建带系统提示的 Agent agent = create_agent( model=chat_model, tools=[weather_tool, calculator_tool], system_prompt="""你是一个友好的助手。特点： - 使用工具前先说明是否需要使用工具，及调用工具的参数 - 结果用表格或列表清晰展示 """, ) print("\n测试：自定义行为的 Agent") response = agent.invoke( { "messages": [ {"role": "user", "content": "北京天气如何？顺便算一下 100 加 50"} ] } ) final_message = response["messages"][-1] print(final_message.content) print("\n关键点：") print(" - system_prompt 定义 Agent 的行为风格") print(" - 可以指定输出格式、语气、工作流程等") # ============================================================================ # 示例 4：Agent 执行过程详解 # ============================================================================ def example_4_agent_execution_details(): """ 示例4：查看 Agent 执行的完整过程 理解 Agent 如何一步步工作 """ agent = create_agent(model=chat_model, tools=[calculator_tool]) response = agent.invoke( {"messages": [{"role": "user", "content": "25 乘以 8 等于多少？"}]} ) # 显示完整的消息历史 print("\n完整消息历史：") for i, msg in enumerate(response["messages"], 1): print(f"\n--- 消息 {i} ({msg.__class__.__name__}) ---") if hasattr(msg, "content"): print(f"内容：{msg.content}") if hasattr(msg, "tool_calls") and msg.tool_calls: print(f"工具调用：{msg.tool_calls}") # ============================================================================ # 示例 5：多轮对话 Agent # ============================================================================ def example_5_multi_turn_agent(): """ 示例5：Agent 的多轮对话 关键：传入历史消息 """ agent = create_agent(model=chat_model, tools=[calculator_tool]) response1 = agent.invoke( {"messages": [{"role": "user", "content": "10 加 5 等于多少？"}]} ) print(f"Agent：{response1['messages'][-1].content}") # 第二轮：继续上一轮的对话 print("\n用户：再乘以 3 呢？") response2 = agent.invoke( { "messages": response1["messages"] + [{"role": "user", "content": "再乘以 3 呢？"}] } ) print(f"Agent：{response2['messages'][-1].content}") # ============================================================================ # 示例 6：Agent 最佳实践 # ============================================================================ def example_6_agent_best_practices(): """ 示例6：Agent 最佳实践 """ # 最佳实践： # 1. 工具选择 # - 只给 Agent 需要的工具（工具太多会混淆） # - 工具的 docstring 要清晰 # - 每个工具功能单一 # 2. System Prompt # - 明确说明 Agent 的角色 # - 定义输出格式 # - 说明何时使用工具 # 3. 错误处理 # - 工具内部捕获异常 # - 返回友好的错误信息 # - Agent 可以处理工具失败 # 4. 性能优化 # - 减少不必要的工具调用 # - 缓存常用查询结果 # - 使用流式输出（后续学习） # 5. 测试 # - 测试各种问题类型 # - 测试边界情况 # - 验证工具选择是否正确 # """) print("\n示例：良好配置的 Agent") agent = create_agent( model=chat_model, tools=[weather_tool, calculator_tool], system_prompt="""你是一个专业的助手。 工作流程： 1. 仔细理解用户问题 2. 如果需要工具，先说明将要做什么 3. 调用工具获取准确信息 4. 基于结果给出清晰答案 输出要求： - 简洁明了 - 数据准确 - 格式清晰 """, ) print("\n测试：") response = agent.invoke( {"messages": [{"role": "user", "content": "北京天气如何？"}]} ) print(f"Agent 回复：{response['messages'][-1].content}") def main(): # example_1_basic_agent() # example_2_multi_tools_agent() # example_3_system_prompt_agent() # example_4_agent_execution_details() # example_5_multi_turn_agent() example_6_agent_best_practices() if __name__ == "__main__": main()
calculator.py""" 自定义工具：计算器 ================== 演示带多个参数的工具 """ from langchain.tools import tool @tool def calculator(operation: str, num1: float, num2: float) -> float: """ 执行基本的数学计算 参数: operation: 运算类型，支持 "add"(加), "subtract"(减), "multiply"(乘), "divide"(除) a: 第一个数字 b: 第二个数字 返回: 计算结果字符串 """ operations = { "add": lambda x, y: x + y, "subtract": lambda x, y: x - y, "multiply": lambda x, y: x * y, "divide": lambda x, y: x / y if y != 0 else "错误：除数不能为零", } if operation not in operations: return f"不支持的运算类型：{operation}。支持的类型：add, subtract, multiply, divide" try: result = operations[operation](num1, num2) return f"{num1} {operation} {num2} = {result}" except Exception as e: return f"计算错误：{e}" # 测试工具 if __name__ == "__main__": print("测试计算器工具：") print(calculator.invoke({"operation": "add", "a": 10, "b": 5})) print(calculator.invoke({"operation": "multiply", "a": 7, "b": 8})) print(calculator.invoke({"operation": "divide", "a": 20, "b": 4})) weather.py""" 自定义工具：天气查询 ==================== 使用 @tool 装饰器创建工具（LangChain 1.0 推荐方式） """ from langchain.tools import tool @tool def get_weather(city: str) -> str: """ 获取指定城市的天气信息 参数: city (str): 要查询天气的城市名称 例如 "北京" 返回: str: 包含天气信息的字符串，格式为 "城市 天气描述 温度" """ # 模拟天气数据 weather_data = { "北京": "晴天，温度 15°C，空气质量良好", "上海": "多云，温度 18°C，有轻微雾霾", "深圳": "阴天，温度 22°C，可能有小雨", "成都": "小雨，温度 12°C，湿度较高", } # 检查城市是否在数据中 if city not in weather_data: return f"城市 {city} 的天气信息未找到" # 返回天气信息 return weather_data.get(city, f"抱歉，暂时没有{city}的天气数据") # 测试工具 if __name__ == "__main__": print("测试天气工具：") print(f"北京天气: {get_weather.invoke({'city': '北京'})}") print(f"上海天气: {get_weather.invoke({'city': '上海'})}") web_search.py""" 自定义工具：网页搜索（模拟） ============================ 演示可选参数的工具 """ from langchain.tools import tool from typing import Optional @tool def web_search(query: str, num_results: Optional[int] = 3) -> str: """ 模拟网页搜索工具 参数: query: 搜索查询 num_results: 返回的结果数量（可选，默认3） 返回: 搜索结果字符串 """ # 模拟搜索结果 mock_results = { "Python": [ "Python官方网站 - https://www.python.org", "Python教程 - 菜鸟教程", "Python最佳实践 - Real Python", ], "机器学习": [ "机器学习入门 - Coursera", "Scikit-learn文档", "机器学习实战 - GitHub", ], "LangChain": [ "LangChain官方文档", "LangChain GitHub仓库", "LangChain教程 - YouTube", ], } # 查找结果 results = [] for key in mock_results: if key.lower() in query.lower(): results.extend(mock_results[key][:num_results]) break if not results: return f"未找到关于'{query}'的结果" # 格式化输出 output = f"搜索 '{query}' 找到 {len(results)} 条结果：\n" for i, result in enumerate(results, 1): output += f"{i}. {result}\n" return output.strip() if __name__ == "__main__": print("测试搜索工具：") print(web_search.invoke({"query": "Python"})) print("\n" + web_search.invoke({"query": "LangChain", "num_results": 2}))
""" LangChain 1.0 - 自定义工具 (@tool 装饰器) ========================================= 本模块重点讲解： 1. 使用 @tool 装饰器创建工具（LangChain 1.0 推荐方式） 2. 工具的参数和文档字符串（docstring）的重要性 3. 测试工具 """ import os import sys from langchain_core.tools import tool from init_model import get_chat_model # 导入自定义工具 from tools import weather from tools import web_search from tools import calculator # ============================================================================ # 示例 1：使用第一个工具 # ============================================================================ def example_1(): """ 示例1：创建天气查询工具 """ # 创建天气查询工具 weather_tool = weather.get_weather # 测试天气工具 print(weather_tool.invoke({"city": "北京"})) print(weather_tool.invoke({"city": "上海"})) def example_2(): tests = [ {"operation": "add", "a": 10, "b": 5}, {"operation": "multiply", "a": 7, "b": 8}, {"operation": "divide", "a": 20, "b": 4}, ] # 创建计算器工具 calc_tool = calculator.get_calculator for test in tests: result = calc_tool.invoke(test) print(f" {result}") def example_3(): """ 工具绑定到模型 """ calc_tool = calculator.get_calculator model_with_tools = get_chat_model(tools=[calc_tool]) response = model_with_tools.invoke("北京今天天气怎么样？") # 检查模型是否要求调用工具 if response.tool_calls: print(f"\n✅ AI 决定使用工具！") print(f"工具调用: {response.tool_calls}") else: print(f"\nℹ️ AI 直接回答（未使用工具）") print(f"回复: {response.content}")
# 04 - Custom Tools (自定义工具) # 核心概念 # 工具 (Tool) = 给 AI 的函数 # 使用 @tool 装饰器，让 AI 能调用你的 Python 函数。 # @tool 基本用法 import json from turtle import st from langchain.tools import tool @tool def get_weather(city: str) -> str: """ 获取指定城市的天气信息 参数: city: 城市名称，如"北京"、"上海" 返回: 天气信息字符串 """ # 你的实现 return "晴天，温度 15°C" # 关键要点 # 必需项 说明 # @tool 装饰器 声明这是一个工具 # docstring AI 读这个来理解工具用途 ⚠️ 非常重要！ # 类型注解 参数和返回值的类型 # 返回 str 工具应该返回字符串（AI 最容易理解） # 工具的 docstring # AI 依赖 docstring 来理解工具！ @tool def my_tool(param: str) -> str: """ 工具的简短描述（AI 读这个！） 参数: param: 参数说明 返回: 返回值说明 """ # 好的 vs 不好的 docstring # ❌ 不好：太模糊 @tool def tool1(x: str) -> str: """做一些事情""" ... # ✅ 好：清晰明确 @tool def search_products(query: str) -> str: """ 在产品数据库中搜索产品 参数: query: 搜索关键词，如"笔记本电脑"、"手机" 返回: 产品列表的 JSON 字符串 """ # 参数类型 # 1. 单参数 # @tool # def get_weather(city: str) -> str: # """获取指定城市的天气""" # ... # 2. 多参数 @tool def calculator(operation: str, a: float, b: float) -> str: """ 执行数学计算 参数: operation: "add", "subtract", "multiply", "divide" a: 第一个数字 b: 第二个数字 """ # 3. 可选参数 from typing import Optional @tool def web_search(query: str, num_results: Optional[int] = 3) -> str: """ 搜索网页 参数: query: 搜索关键词 num_results: 返回结果数量，默认 3 """ # 调用工具 # 工具有两种调用方式： # 1. 直接调用（测试用） # result = get_weather.invoke({"city": "北京"}) # print(result) # "晴天，温度 15°C" from init_model import get_chat_model chat_model = get_chat_model() # 2. 绑定到模型（让 AI 调用） model_with_tools = chat_model.bind_tools([get_weather]) # # AI 可以决定是否调用工具 # response = model_with_tools.invoke("北京天气如何？") # # 检查 AI 是否要调用工具 # if response.tool_calls: # print("AI 想调用工具：", response.tool_calls) # # print("AI 调用工具结果：", response.additional_kwargs["function_call"]["arguments"]) # else: # print("AI 直接回答：", response.content) # 工具属性 # 创建工具后，可以查看其属性： @tool def my_tool(param: str) -> str: """工具描述""" ... print(my_tool.name) # "my_tool" print(my_tool.description) # "工具描述" print(my_tool.args) # 参数模式 # 最佳实践 # 1. 清晰的描述 @tool def search_flights(origin: str, destination: str, date: str) -> str: """ 在航班数据库中搜索航班 参数: origin: 出发地,如"北京" destination: 目的地 如"上海" date: 日期，格式 "YYYY-MM-DD" 返回: 航班列表的 JSON 字符串 """ # 2. 功能单一 # ❌ 不好：一个工具做太多事 # @tool # def do_everything(action: str, data: str) -> str: # """做各种事情""" # if action == "weather": # ... # elif action == "calculate": # ... # elif action == "search": # ... # # ✅ 好：每个工具做一件事 # @tool # def get_weather(city: str) -> str: # """获取天气""" # ... # @tool # def calculator(operation: str, a: float, b: float) -> str: # """计算""" # ... # 3. 错误处理 @tool def divide(a: float, b: float) -> str: """ 除法计算 参数: a: 被除数 b: 除数 """ try: if b == 0: return "错误：除数不能为零" result = a / b return f"{a} / {b} = {result}" except Exception as e: return f"计算错误：{e}" # 4. 返回字符串 @tool def get_user_info(user_id: str) -> str: """获取用户信息""" user = {"id": user_id, "name": "张三"} return json.dumps(user, ensure_ascii=False)
# 03 - Messages: 消息类型与对话管理 # 核心要点（只讲难点） # 初始化模型 from init_model import get_chat_model chat_model = get_chat_model() # 角色 字典格式 对象格式 用途 # System {"role": "system", ...} SystemMessage(...) 系统提示 # User {"role": "user", ...} HumanMessage(...) 用户输入 # Assistant {"role": "assistant", ...} AIMessage(...) AI 回复 # 推荐：直接用字典，简洁！ def demo1(): message = [ {"role": "system", "content": "你是一个专业的编程导师，擅长解释复杂概念。"}, {"role": "user", "content": "什么是列表？"}, ] # ❌ 不推荐（太啰嗦） from langchain_core.messages import SystemMessage, HumanMessage messages = [SystemMessage(content="你是助手"), HumanMessage(content="你好")] # 2. 对话历史管理（核心难点） # 🔴 关键规则 # 每次调用必须传递完整的对话历史！ def demo2(): # 第一次 r1 = chat_model.invoke("我叫张三") # 第二次（没传历史） r2 = chat_model.invoke("我叫什么？") # AI 不记得！ # ✅ 正确做法 def demo3(): conversation = [] conversation.append({"role": "user", "content": "我叫张三"}) r1 = chat_model.invoke(conversation) conversation.append({"role": "assistant", "content": r1.content}) # 第二次（传递完整历史） conversation.append({"role": "user", "content": "我叫什么？"}) r2 = chat_model.invoke(conversation) # AI 记得！ # 3. 对话历史优化（避免太长） # 🔴 问题 # 对话历史会越来越长，消耗大量 tokens 和成本。 # ✅ 解决方案 # 只保留最近 N 轮对话： def keep_recent_messages(messages, max_pairs=2): """ 保留最近的 N 轮对话 max_pairs: 保留的对话轮数（每轮 = user + assistant） """ if not messages: return [] # 1. 处理 System 消息：通常只保留第一条作为系统指令 # 如果你的业务逻辑需要保留所有 system 消息，可以改回原来的列表推导式 system_msg = next((m for m in messages if m.get("role") == "system"), None) # 2. 提取纯对话流 (排除 system) conversation = [m for m in messages if m.get("role") != "system"] if not conversation: return [system_msg] if system_msg else [] # 3. 智能截取最近对话 # 我们需要保留最多 max_pairs 轮。 # 一轮定义为：User -> Assistant # 特殊情况：最后一条如果是 User，也要保留（即使没有对应的 Assistant） recent_conversation = [] pairs_found = 0 # 从后往前遍历，每次找到 User 就添加它和对应的 Assistant（如果有） i = len(conversation) - 1 while i >= 0 and pairs_found < max_pairs: current_msg = conversation[i] # 如果是 Assistant 消息 if current_msg.get("role") == "assistant": # 尝试找前一条是否为 User if i > 0 and conversation[i - 1].get("role") == "user": # 找到完整的一对 recent_conversation.insert(0, conversation[i - 1]) recent_conversation.insert(1, current_msg) pairs_found += 1 i -= 2 else: # 孤立的 Assistant 消息（异常情况），单独保留或跳过？ # 这里选择保留它以防丢失上下文，但不计入完整轮数 recent_conversation.insert(0, current_msg) i -= 1 else: # 如果是 User 消息 # 情况 A: 它是最后一条消息（当前正在提问），必须保留 # 情况 B: 它是某轮的开始，但后面没有 Assistant（数据损坏），保留 # 只要还没凑够 max_pairs，且这是剩下的最后一条 User，就保留 if pairs_found < max_pairs: recent_conversation.insert(0, current_msg) i -= 1 # 4. 组装结果 result = [] if system_msg: result.append(system_msg) result.extend(recent_conversation) return result conversation_history = [ {"role": "system", "content": "你是一个助手"}, {"role": "user", "content": "你好"}, {"role": "assistant", "content": "你好！有什么事吗？"}, {"role": "user", "content": "今天天气如何"}, {"role": "assistant", "content": "天气不错"}, {"role": "user", "content": "那明天呢？"}, # 当前最新的问题，尚未回答 ] # 使用 optimized = keep_recent_messages(conversation_history, max_pairs=5) response = chat_model.invoke(optimized) # 核心总结 # 要点 说明 # 格式 用字典，不用消息对象 # 历史 每次必须传递完整历史 # 保存 必须保存 AI 的回复 # 优化 只保留最近 N 轮 # System 总是保留 system 消息
""" LangChain 1.0 基础教程 - 提示词模板 (Prompt Templates) ======================================================== 本文件演示如何使用 LangChain 的提示词模板系统 涵盖以下核心概念： 1. PromptTemplate - 简单文本模板 2. ChatPromptTemplate - 聊天消息模板 3. 模板变量和格式化 4. 消息模板的组合 5. 实际应用场景 作者：LangChain 学习者 日期：2026年3月9日 """ from openai import chat from init_model import get_chat_model from langchain_core.prompts import ( PromptTemplate, ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, AIMessagePromptTemplate, ) # 初始化聊天模型 代码见init_model.py chat_model = get_chat_model() # ============================================================================ # 示例 1：为什么需要提示词模板？ # ============================================================================ def example_1(): """ 示例1：对比字符串拼接 vs 模板 问题：字符串拼接容易出错、难维护、不可复用 解决：使用提示词模板 """ print("\n" + "=" * 70) print("示例 1：为什么需要提示词模板？") print("=" * 70) # ❌ 不推荐：使用字符串拼接 print("\n【方式 1：字符串拼接（不推荐）】") print("-" * 70) topic = "Python" difficulty = "初学者" # 难以维护，容易出错 prompt_str = f"你是一个{difficulty}级别的编程导师。请用简单易懂的语言解释{topic}。" print(f"提示词：{prompt_str}") response = chat_model.invoke(prompt_str) print(f"AI 回复：{response.content[:100]}...\n") # ✅ 推荐：使用 PromptTemplate template = PromptTemplate.from_template( "你是一个{difficulty}级别的编程导师。请用简单易懂的语言解释{topic}。" ) print(f"模板：{template.template}") print(f"变量：{template.input_variables}") prompt = template.format(difficulty=difficulty, topic=topic) response = chat_model.invoke(prompt) print(f"AI 回复：{response.content[:100]}...\n") print("💡 优势：") print(" 1. 可复用 - 同一个模板可以用于不同的输入") print(" 2. 可维护 - 模板和数据分离，易于修改") print(" 3. 类型安全 - 自动验证变量") print(" 4. 可测试 - 更容易编写测试用例") # 示例 2：PromptTemplate 基础用法 def example_2(): """ 示例2：PromptTemplate 的基本用法 PromptTemplate 用于简单的文本模板 适合单一提示词的场景 """ print("\n" + "=" * 70) print("示例 2：PromptTemplate 基础用法") print("=" * 70) # 方法 1：使用 from_template（最简单） print("\n【方法 1：from_template（推荐）】") template1 = PromptTemplate.from_template("将以下文本翻译成{language}：\n{text}") prompt1 = template1.format(language="法语", text="Hello, how are you?") print(f"提示词 1：{prompt1}") response1 = chat_model.invoke(prompt1) print(f"AI 回复 1：{response1.content[:100]}...\n") # 方法 2：显式指定变量（更严格） print("【方法 2：显式指定变量】") template2 = PromptTemplate( input_variables=["product", "feature"], template="为{product}写一句广告语，重点突出{feature}特点。", ) prompt2 = template2.format(product="智能音箱", feature="语音识别") print(f"提示词 2：{prompt2}") response2 = chat_model.invoke(prompt2) print(f"AI 回复：{response2.content}\n") # 方法 3：使用 invoke（直接生成消息） print("【方法 3：使用 invoke（更方便）】") template3 = PromptTemplate.from_template( "写一首关于{theme}的{style}风格的诗，不超过4行。" ) # invoke 直接返回格式化后的值 prompt_value = template3.invoke({"theme": "春天", "style": "现代"}) def example_3(): """ 示例3：ChatPromptTemplate 的基本用法 ChatPromptTemplate 用于构建聊天消息 支持 system、user、assistant 多种角色 """ # 方法 1：使用元组格式（最简单，推荐） print("\n【方法 1：元组格式】") chat_template = ChatPromptTemplate.format_messages( [ ("system", "你是一个{role}，擅长{expertise}。"), ("user", "请解释{topic}。"), ] ) # 格式化模板 messages = chat_template.format_messages( role="Python 导师", expertise="用简单的方式解释复杂概念", task="解释什么是列表推导式", ) response = chat_model.invoke(messages) # 方法 2：使用字符串简写（最简洁） simple_template = ChatPromptTemplate.from_messages( [("system", "你是一个友好的助手"), ("user", "{question}")] ) messages = simple_template.format_messages(question="什么是机器学习？") response = chat_model.invoke(messages) # ============================================================================ # 示例 4：多轮对话模板 # ============================================================================ def example_4(): """ 示例4：构建多轮对话的模板 包含系统提示、对话历史和当前问题 """ print("\n" + "=" * 70) print("示例 4：多轮对话模板") print("=" * 70) # 创建包含对话历史的模板 template = ChatPromptTemplate.from_messages( [ ("system", "你是一个专业的编程导师，擅长解释复杂概念。"), ("user", "{question}"), ("assistant", "{answer1}"), ("user", "{question2}"), ] ) print("模板结构：") print(" 1. System: 设定角色和指令") print(" 2. User: 第一个问题") print(" 3. Assistant: 第一个回答") print(" 4. User: 第二个问题（基于上下文）\n") # 填充模板 prompt = template.format_messages( role="Python 专家", instruction="回答要简洁、准确", question="什么是列表？", answer1="列表是 Python 中的有序可变集合，用方括号 [] 表示。", question2="它和元组有什么区别？", # 基于上下文的问题 ) response = chat_model.invoke(prompt) # ============================================================================ # 示例 5：使用 MessagePromptTemplate（高级） # ============================================================================ def example_5(): """ 示例5：使用 MessagePromptTemplate 类 提供更细粒度的控制 """ print("\n" + "=" * 70) print("示例 5：MessagePromptTemplate 类（高级用法）") print("=" * 70) # 分别创建不同类型的消息模板 system_template = SystemMessagePromptTemplate.from_template( "你是一个{profession}，你的特长是{specialty}。" ) human_template = HumanMessagePromptTemplate.from_template("请解释{topic}。") ai_template = AIMessagePromptTemplate.from_template("我会用{language}来解释。") # 组合成 ChatPromptTemplate chat_template = ChatPromptTemplate.from_messages([system_template, human_template]) # 使用模板 messages = chat_template.format_messages( profession="数据科学家", specialty="用数据讲故事", topic="数据可视化", question="如何选择合适的图表类型？", ) response = chat_model.invoke(messages) print(f"AI 回复：{response.content[:200]}...\n") # ============================================================================ # 示例 6：部分变量（Partial Variables） # ============================================================================ def example_6(): """ 示例6：部分变量 - 预填充某些变量 适用场景： - 某些变量固定不变 - 需要创建模板变体 """ print("\n" + "=" * 70) print("示例 6：部分变量（Partial Variables）") print("=" * 70) # 创建原始模板 # 创建原始模板 original_template = ChatPromptTemplate.from_messages( [("system", "你是一个{role}，你的目标用户是{audience}。"), ("user", "请{task}")] ) # 部分填充：固定 role 和 audience partially_filled = original_template.partial(role="科技博客作者", audience="程序员") messages1 = partially_filled.format_messages( task="写一篇关于 Python 装饰器的文章开头" ) response1 = chat_model.invoke(messages1) print(f"文章 1：{response1.content[:150]}...\n") # 复用模板，不同的 task messages2 = partially_filled.format_messages(task="写一篇关于异步编程的文章开头") response2 = model.invoke(messages2) print(f"文章 2：{response2.content[:150]}...\n") def example_7(): """ 示例9：模板 + 模型的链式调用 LangChain Expression Language (LCEL) """ print("\n" + "=" * 70) print("示例 9：LCEL 链式调用（预览）") print("=" * 70) # 创建模板 template = ChatPromptTemplate.from_messages( [("system", "你是一个{role}"), ("user", "{input}")] ) # 使用 | 运算符创建链 chain = template | chat_model print("链的组成：") print(" 模板 | 模型") print(" (Template) | (LLM)\n") # 直接调用链 response = chain.invoke({"role": "幽默的程序员", "input": "解释什么是bug"}) print(f"AI 回复：{response.content}\n") print("💡 链式调用的优势：") print(" 1. 代码更简洁") print(" 2. 组件可复用") print(" 3. 易于调试和监控") print(" （详细内容将在后续模块学习）") def main(): example_1() input("\n按 Enter 继续...") example_2() input("\n按 Enter 继续...") example_3() input("\n按 Enter 继续...") example_4() input("\n按 Enter 继续...") example_5() input("\n按 Enter 继续...") example_6() input("\n按 Enter 继续...") if __name__ == "__main__": main()
# =================================== # 学习目标 # =================================== # 通过本模块，你将学习： # 1. 为什么需要提示词模板 # 字符串拼接的问题 # 模板的优势 # 2.PromptTemplate # 基本用法 # 变量替换 # 格式化方法 # 3.ChatPromptTemplate # 聊天消息模板 # 多角色支持 # 对话历史管理 # 4.高级特性 # 部分变量 # 模板组合 # 可复用模板库 # 5.LCEL 链式调用 # 模板与模型的组合 # 管道运算符 from regex import D, P from init_model import get_chat_model chat_model = get_chat_model() # 使用模板 from langchain_core.prompts import ChatPromptTemplate, PromptTemplate # 1. 为什么需要提示词模板？ # 🔴 问题：字符串拼接的缺点 # ❌ 不推荐的做法 def demo1(): user_name = "张三" topic = "Python" prompt = f"你好 {user_name}，我来帮你学习 {topic}" # 问题： # ❌ 难以维护和修改 # ❌ 容易出现格式错误 # ❌ 不能复用 # ❌ 难以测试 # ❌ 混合了逻辑和数据 def demo2(): template = PromptTemplate.from_template("你好 {user_name}，我来帮你学习 {topic}") prompt = template.format(user_name="张三", topic="Python") # ✅ 可复用 - 一个模板，多次使用 # ✅ 可维护 - 模板和数据分离 # ✅ 类型安全 - 自动验证变量 # ✅ 可测试 - 更容易编写测试 # ✅ 可组合 - 可以组合多个模板 # 2. PromptTemplate - 简单文本模板 # PromptTemplate 用于创建简单的文本提示词，适合单一提示的场景。 # 方法 1：from_template（最简单，推荐） def demo3(): template = PromptTemplate.from_template("你的模板文本 {变量名}") # 方法 2：完整定义 template = PromptTemplate( input_variables=["变量1", "变量2"], template="你的模板文本 {变量1} 和 {变量2}" ) ##创建模板的三种方法 # 方法 1：from_template（推荐） def demo4_1(): template = PromptTemplate.from_template("将以下文本翻译成{language}：\n{text}") # 自动识别变量 print(template.input_variables) # ['language', 'text'] def demo4_2(): template = PromptTemplate( input_variables=["product", "feature"], template="为{product}写一句广告语，重点突出{feature}特点。", ) # 方法 部分变量预填充 def demo4_3(): template = PromptTemplate.from_template("你是一个{role}，请{task}") # 预填充 role partial_template = template.partial(role="Python 导师") # 填充 task prompt = partial_template.format(task="解释一下什么是大模型？") print(prompt) # 你是一个Python 导师，解释一下什么是大模型？ # 使用模板方法 # 方式 1：format() - 返回字符串 def demo5_1(): temp = PromptTemplate.from_template("你好{name}") # 返回格式化后的字符串 prompt_str = temp.format(name="张三") print(prompt_str) # 你好张三 # 直接传递给模型 response = chat_model.invoke(prompt_str) print(response.content) # 你好张三，我来帮你学习 Python # 方式 2：invoke() - 返回 PromptValue def demo5_2(): temp = PromptTemplate.from_template("你好{name}") # 返回 PromptValue 对象 prompt_value = temp.invoke({"name": "张三"}) # 获取文本 print(prompt_value.text) # "你好 张三" # =================================== # 模版实战案例 # =================================== # 示例 1：翻译模板 def demo6(): translator = PromptTemplate.from_template( "将以下{source_lang}文本翻译成{target_lang}：\n{text}" ) promat = translator.format( source_lang="英文", target_lang="中文", text="Hello, how are you?" ) # 示例 2：代码生成模板】、 def demo7(): code_template = PromptTemplate.from_template( "用{language}编写一个{functionality}的函数。\n" "要求：\n" "1. {requirement1}\n" "2. {requirement2}" ) prompt = code_template.format( language="Python", functionality="计算斐波那契数列", requirement1="使用递归实现", requirement2="添加类型注解", ) # 3. ChatPromptTemplate - 聊天消息模板 # ChatPromptTemplate 用于创建聊天格式的消息，支持多种角色（system、user、assistant）。 # 为什么需要 ChatPromptTemplate？ # PromptTemplate vs ChatPromptTemplate： # 特性 PromptTemplate ChatPromptTemplate # 输出格式 纯文本字符串 消息列表 # 角色支持 ❌ 无 ✅ system/user/assistant # 对话历史 ❌ 不支持 ✅ 支持 # 适用场景 简单提示 聊天、对话、多轮交互 # 消息类型 # 角色字符串 含义 用途 # "system" 系统消息 设定 AI 的行为、角色、规则 # "user" / "human" 用户消息 用户的输入/问题 # "assistant" / "ai" AI 消息 AI 的回复（用于对话历史） # 创建方法 # 方法 1：元组格式（最简单，推荐） def demo8_1(): chat_template = ChatPromptTemplate.from_messages( [("system", "你是一个专业的翻译。"), ("user", "{input}")] ) messages = chat_template.format_messages( role="Python 导师", question="什么是装饰器？" ) # 方法 2：字符串简写 # 单独的字符串会被解释为 user 消息 def demo8_2(): template = ChatPromptTemplate.from_messages( [("system", "你是助手"), "{user_input}"] # 相当于 ("user", "{user_input}") ) messages = template.format_messages(user_input="你好") # 方法 3：使用 MessagePromptTemplate（高级） from langchain_core.prompts import ( SystemMessagePromptTemplate, HumanMessagePromptTemplate, ) def demo8_3(): system_template = SystemMessagePromptTemplate.from_template("你是一个{role}") human_template = HumanMessagePromptTemplate.from_template("{question}") template = ChatPromptTemplate.from_messages([system_template, human_template]) messages = template.format_messages(role="Python 导师", question="什么是装饰器？") ## 4. 模板的使用场景 # 方式 1：format_messages() - 返回消息列表 def demo9_1(): template = ChatPromptTemplate.from_messages( [("system", "你是{role}"), ("user", "{input}")] ) message = template.format_messages(role="Python 导师", input="什么是装饰器？") # 直接传递给模型 response = chat_model.invoke(message) print( response.content ) # 装饰器是 Python 中一种特殊的函数，用于修改其他函数的行为。 # 方式 2：invoke() - 返回 ChatPromptValue def demo9_2(): template = ChatPromptTemplate.from_messages( [("system", "你是{role}"), ("user", "{input}")] ) # 返回 ChatPromptValue 对象 prompt_value = template.invoke({"role": "助手", "input": "你好"}) messages = prompt_value.to_messages() print(messages) # =================================== # 实用案例 # =================================== # 示例 1：简单聊天 def demo_chat(): template = ChatPromptTemplate.from_messages( [("system", "你是一个友好的{role}，擅长{skill}"), ("user", "{question}")] ) messages = template.format_messages( role="编程导师", skill="用简单语言解释复杂概念", question="什么是递归？" ) response = chat_model.invoke(messages) # 示例 2：多轮对话 def demo_chat_multi(): template = ChatPromptTemplate.from_messages( [ ("system", "你是一个专业的{role}，擅长{skill}"), ("user", "{question}"), ("assistant", "{answer}"), ] ) messages = template.format_messages( role="Python 专家", question1="什么是列表？", answer1="列表是 Python 的有序可变集合。", question2="它和元组有什么区别？", # 基于上下文 ) # 示例 3：结构化指令 def demo_chat1(): structured_template = ChatPromptTemplate.from_messages( [ ( "system", "你是一个{domain}专家。\n" "回答风格：{style}\n" "回答长度：{length}字以内", ), ("user", "{question}"), ] ) messages = structured_template.format_messages( domain="机器学习", style="技术性强、简洁", length="100", question="什么是梯度下降？", ) # 4. 高级特性 # 4.1 部分变量（Partial Variables） # 预填充某些固定不变的变量，创建模板的变体 # 使用场景： # 某些变量在所有调用中都相同 # 需要为不同用户/场景创建定制模板 # 语法： def demo_partial(): # 创建一个初始模板 template = ChatPromptTemplate.from_messages( [("system", "你是{role}，目标用户是{audience}"), ("user", "{task}")] ) # 预填充模版部分数据 customer_support_template = template.partial(role="客服专员", audience="普通用户") # 现在只需要提供 task messages = customer_support_template.format_messages(task="解释退款政策") response = chat_model.invoke(messages) # 实用示例： def demo_chat2(): # 创建一个初始模板 # 基础翻译模板 translator = ChatPromptTemplate.from_messages( [("system", "你是专业翻译，精通{source}和{target}"), ("user", "翻译：{text}")] ) # 创建英译中的专用模板 en_to_zh = translator.partial(source="英语", target="中文") # 创建中译英的专用模板 zh_to_en = translator.partial(source="中文", target="英语") # 英译中 messages = en_to_zh.format_messages(text="Hello, how are you?") response = chat_model.invoke(messages) print(response.content) # 中译英 messages = zh_to_en.format_messages(text="你好，我很好") response = chat_model.invoke(messages) print(response.content) # 模板组合 # 将多个模板片段组合成复杂的提示词。 # 方法 1：字符串组合 def demo_chat3(): # 定义可复用的部分 # 定义可复用的部分 role_part = "你是一个{domain}专家。" style_part = "回答风格：{style}。" constraint_part = "限制：{constraint}。" # 组合 full_system = role_part + style_part + constraint_part template = ChatPromptTemplate.from_messages( [("system", full_system), ("user", "{question}")] ) # 方法 2：使用 + 运算符 def demo_chat4(): template1 = ChatPromptTemplate.from_messages([("system", "你是助手")]) template2 = ChatPromptTemplate.from_messages([("user", "{input}")]) # 组合（LangChain 1.0 支持） combined = template1 + template2 # 格式化 messages = combined.format_messages(input="你好") response = chat_model.invoke(messages) print(response.content) # 4.3 可复用模板库 # 在实际项目中，建议创建模板库。 # 示例：模板库 # templates.py # from langchain_core.prompts import ChatPromptTemplate # class PromptTemplates: # """可复用的提示词模板库""" # TRANSLATOR = ChatPromptTemplate.from_messages( # [("system", "你是专业翻译，精通{source}和{target}"), ("user", "翻译：{text}")] # ) # CODE_REVIEWER = ChatPromptTemplate.from_messages( # [ # ("system", "你是{language}代码审查专家，重点关注{focus}"), # ("user", "审查代码：\n```{language}\n{code}\n```"), # ] # ) # SUMMARIZER = ChatPromptTemplate.from_messages( # [ # ("system", "你是内容摘要专家"), # ("user", "将以下内容总结为{num}个要点：\n{content}"), # ] # ) # TUTOR = ChatPromptTemplate.from_messages( # [("system", "你是{subject}导师，学生水平：{level}"), ("user", "{question}")] # ) # 使用 # from templates import PromptLibrary # messages = PromptLibrary.TRANSLATOR.format_messages( # source_lang="英语", # target_lang="中文", # text="Hello World" # ) # 5. LCEL 链式调用（预览） # LCEL = LangChain Expression Language，LangChain 的表达式语言。 # 什么是链（Chain）？ # 链是将多个组件连接在一起的方式，形成处理流程。 # 输入 → 模板 → 模型 → 输出 # 使用管道运算符 | def demo_chat5(): # 创建组件 template = ChatPromptTemplate.from_messages( [("system", "你是{role}"), ("user", "{input}")] ) # 使用 | 创建链 chain = template | chat_model response = chain.invoke( { "role": "客服专员", "input": "解释退款政策", } ) print(response.content) # 链的优势 # 优势 说明 # 简洁 一行代码完成多步操作 # 可读 清晰展示数据流向 # 可组合 可以轻松添加/删除组件 # 可复用 链本身可以作为组件 # 详细内容将在后续模块学习。 # 常见问题 (FAQ) # Q1: PromptTemplate 和 ChatPromptTemplate 有什么区别? # A: # 特性 PromptTemplate ChatPromptTemplate # 输出 字符串 消息列表 # 角色 无 system/user/assistant # 适用场景 简单提示 聊天、对话 # 建议： # 简单场景 → PromptTemplate # 聊天场景 → ChatPromptTemplate（推荐） # Q2: 什么时候使用部分变量？ # A: 当某些变量在多次调用中保持不变时： # 场景：为不同部门创建专用模板\ # def demo_chat6(): # base_template = ChatPromptTemplate.from_messages( # [("system", "你是{department}的{role}"), ("user", "{task}")] # ) # # IT 部门 # it_template = base_template.partial(department="IT", role="开发工程师") # # 销售部门 # sales_template = base_template.partial(department="销售部门", role="销售顾问") # Q3: 如何在模板中使用换行和特殊字符？ # A: 使用三引号字符串： def demo_chat7(): template = PromptTemplate.from_template( """ 你是一个{role}。 请完成以下任务： 1. {task1} 2. {task2} 注意事项： - {note1} - {note2} """ ) # Q4: 模板变量可以是什么类型？ # A: 通常是字符串，但也可以是其他可转换为字符串的类型： def demo_chat8(): template = PromptTemplate.from_template("生成{count}个关于{topic}的想法") # count 是整数 prompt = template.format(count=5, topic="创新") # Q5: 如何处理可选变量？ # A: 使用部分变量或默认值： def demo_chat9(): # 方法 1：部分变量 template = PromptTemplate.from_template("{greeting} {name}，{message}") template_with_default = template.partial(greeting="你好") # 方法 2：在应用层处理 def create_prompt(name, message, greeting="你好"): return template.format(greeting=greeting, name=name, message=message) # 最佳实践 # 1. 模板命名规范 # ✅ 好的命名 # translator_template = ... # code_review_template = ... # customer_support_template = ... # # ❌ 不好的命名 # template1 = ... # t = ... # my_template = ... # 2.组织模版 # templates/ # ├── __init__.py # ├── common.py # 通用模板 # ├── translation.py # 翻译相关 # └── coding.py # 编程相关 def demo_chat9_2(): # common.py FRIENDLY_ASSISTANT = ChatPromptTemplate.from_messages( [("system", "你是一个友好的助手"), ("user", "{input}")] ) # 3. 文档化模板 class Templates: """项目模板库""" TRANSLATOR = ChatPromptTemplate.from_messages( [("system", "你是专业翻译"), ("user", "翻译：{text}")] ) """ 翻译模板 变量: - text: 要翻译的文本 示例: messages = TRANSLATOR.format_messages(text="Hello") """ # 4. 测试模板 # def test_translator_template(): # """测试翻译模板""" # template = PromptLibrary.TRANSLATOR # # 测试变量识别 # assert "text" in template.input_variables # # 测试格式化 # messages = template.format_messages(text="Hello") # assert len(messages) == 2 # assert messages[0].type == "system" # assert messages[1].type == "user" def main(): demo9_2() if __name__ == "__main__": main()