乘风的人 - Python Code
type: channel
link: https://blog.zhangbiechen.cn/index.php/tag/Python-Code
items: Mac 使用transformers 库运行Qwen3-Embedding-0.6B模型使用全记录
items: RAG - 文档加载器
items: Validation & Retry.py
items: Structured Output.py
items: Middleware Basics -main.py
items: Middleware Basics.py
items: Checkpointing - main.py
items: check_point.py
items: Context Management - main.py
items: Memory Basics - main.py
items: Memory Basics.py
items: Agent Loop - main.py
items: Agent Loop.py
items: Simple Agent -main(创建 Agent调用工具).py
items: tools 工具类
items: Custom Tools - main.py
items: Custom_Tools.py
items: messages.py
items: 提示词模版 - main.py
items: 提示词模板-Prompt_Templates.py

Mac 使用transformers 库运行Qwen3-Embedding-0.6B模型使用全记录
type: item
date: 2026-03-30T14:10:00+08:00
description: 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. 配置区域 (请在此处修改本地模型路径) # ==…

RAG - 文档加载器
type: item
date: 2026-03-19T17:08:00+08:00
description: 加载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, # "p…

Validation & Retry.py
type: item
date: 2026-03-17T15:20:24+08:00
description: # 核心概念 # 验证和重试 = 确保 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_mo…

Structured Output.py
type: item
date: 2026-03-17T11:53:19+08:00
description: # 核心概念 # 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(da…

Middleware Basics -main.py
type: item
date: 2026-03-17T09:40:39+08:00
description: """ 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.war…

Middleware Basics.py
type: item
date: 2026-03-17T09:39:51+08:00
description: # 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…

Checkpointing - main.py
type: item
date: 2026-03-13T15:16:01+08:00
description: """ 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.…

check_point.py
type: item
date: 2026-03-13T15:15:17+08:00
description: # 核心概念 # 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…

Context Management - main.py
type: item
date: 2026-03-12T09:27:00+08:00
description: """ 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 …

Memory Basics - main.py
type: item
date: 2026-03-11T14:15:00+08:00
description: """ 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()

Memory Basics.py
type: item
date: 2026-03-11T14:14:48+08:00
description: # 核心概念 # 内存 = 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_alic…

Agent Loop - main.py
type: item
date: 2026-03-11T09:57:00+08:00
description: """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,…

Agent Loop.py
type: item
date: 2026-03-11T09:56:47+08:00
description: # 核心概念 # 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…

Simple Agent -main(创建 Agent调用工具).py
type: item
date: 2026-03-10T17:21:47+08:00
description: 目录结构 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 等于多…

tools 工具类
type: item
date: 2026-03-10T15:01:00+08:00
description: 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教程 - 菜鸟教程", "P…

Custom Tools - main.py
type: item
date: 2026-03-10T14:59:18+08:00
description: """ 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}")

Custom_Tools.py
type: item
date: 2026-03-10T14:57:20+08:00
description: # 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: 出…

messages.py
type: item
date: 2026-03-09T14:34:49+08:00
description: # 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": #…

提示词模版 - main.py
type: item
date: 2026-03-09T10:54:00+08:00
description: """ 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 ar…

提示词模板-Prompt_Templates.py
type: item
date: 2026-03-09T09:56:00+08:00
description: # =================================== # 学习目标 # =================================== # 通过本模块，你将学习： # 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) # 你好张三，我来帮你学习 Pytho…