📊 核心流程图（Mermaid 示意）核心在于 review 审核不通过时会指向 revise（修改），而 revise 执行完又会指回 review，形成闭环。graph TD; START([开始]) --> apply[提交贷款申请]; apply --> review[风控审核]; review -->|"✅ 批准"| approve[放款成功] --> END([结束]); review -->|"❌ 拒绝（评分过低）"| reject[直接拒绝] --> END; %% 核心循环：返回上一层（退回修改） review -->|"🔄 补充材料/信息有误"| revise[申请人修改补充]; revise -->|"修改完成，再次提交"| review; style revise fill:#f9f,stroke:#333,stroke-width:2px; style review fill:#bbf,stroke:#333,stroke-width:2px;💻 完整可运行代码（带循环返回机制）这个代码模拟了贷款审批的完整闭环，你复制运行就能看到状态如何在 审核 和 修改 之间反复横跳。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 定义状态（记录贷款信息和循环次数） class LoanState(TypedDict): applicant: str income: int # 月收入（万） credit_score: int # 信用分 (0-100) missing_docs: str # 缺失材料描述 attempts: int # 当前是第几次提交审核 status: str # "pending", "approved", "rejected" # 2. 节点 A：提交申请（初始化或修改后重新提交） def submit_application(state: LoanState): print(f"📄 [{state['applicant']}] 提交申请，收入={state['income']}万，信用分={state['credit_score']}") # 每次提交，尝试次数 +1 return {"attempts": state.get("attempts", 0) + 1} # 3. 节点 B：风控审核（核心判断节点） def review_application(state: LoanState): income = state["income"] score = state["credit_score"] attempts = state["attempts"] missing = state.get("missing_docs", "") print(f"🔍 第 {attempts} 次审核 {state['applicant']}...") # 业务规则判断 if score < 50: return {"status": "rejected"} # 信用太差，直接拒绝，不循环 if income < 5: # 【核心循环触发点】：收入证明不足，退回修改 return { "status": "pending", "missing_docs": "请补充近6个月银行流水（当前收入低于5万）" } if missing != "": # 如果之前有缺失材料，审核再次发现没补全，继续退回 return { "status": "pending", "missing_docs": "材料仍未补全，请重新上传身份证复印件" } # 全部通过 return {"status": "approved"} # 4. 节点 C：修改补充材料（返回上一层后的处理动作） def revise_documents(state: LoanState): print(f"✏️ {state['applicant']} 正在修改材料：{state['missing_docs']}") # 模拟申请人修改：提高收入（编造数据模拟修正），并清空缺失提示 # 注意：这里修改状态，使得再次进入 review 时能通过 return { "income": state["income"] + 3, # 补交材料证明实际收入更高 "missing_docs": "", # 清空错误标记 "attempts" : state["attempts"] + 1 } # 5. 【核心】条件路由：决定下一步去哪儿（是实现"返回上一层"的关键） def route_after_review(state: LoanState) -> Literal["approve", "reject", "revise"]: status = state["status"] attempts = state["attempts"] # 安全熔断：如果循环超过 3 次，强制拒绝，防止死循环 if attempts >= 3: print("⚠️ 超过最大修改次数（3次），强制拒绝！") return "reject" if status == "approved": return "approve" elif status == "rejected": return "reject" else: # pending 状态，需要修改 return "revise" # 指向修改节点（这就是返回上一层） # 6. 构建图 builder = StateGraph(LoanState) # 添加节点 builder.add_node("submit", submit_application) builder.add_node("review", review_application) builder.add_node("revise", revise_documents) # 定义终点动作节点（为了流程图清晰，单独列出来） def approve_loan(state): print(f"✅ 贷款已批准！恭喜 {state['applicant']}") return state def reject_loan(state): print(f"❌ 贷款被拒绝。{state['applicant']} 信用分 {state['credit_score']}") return state builder.add_node("approve", approve_loan) builder.add_node("reject", reject_loan) # 设置边（流转逻辑） builder.add_edge(START, "submit") builder.add_edge("submit", "review") # 【核心循环边】：review 根据路由结果，可能指向 revise（返回上一级） builder.add_conditional_edges("review", route_after_review, { "revise": "revise", "approve": "approve", "reject": "reject" }) # 【闭环边】：revise（修改节点）执行完后，重新指回 review（审核节点），实现循环 builder.add_edge("revise", "review") # 终点 builder.add_edge("approve", END) builder.add_edge("reject", END) # 7. 编译并运行测试 graph = builder.compile() # 打印 ASCII 流程图，查看循环结构 print("=" * 30 + " 流程图结构 " + "=" * 30) print(graph.get_graph().draw_mermaid()) print("=" * 80) # 模拟一个需要修改2次才能通过的申请人 initial_state = { "applicant": "张三", "income": 4, # 低于5万，会触发第一次退回 "credit_score": 70, # 信用良好 "missing_docs": "", "attempts": 0, "status": "pending" } print("\n🚀 开始贷款审批流程...\n") final_state = graph.invoke(initial_state) print("\n📌 最终状态:", final_state)🎯 执行结果演示（你会看到循环过程）运行上面的代码，控制台会输出：================================ 流程图结构 ================================ +-----------+ +--------+ +--------+ +----------+ +--------+ | __start__ | --> | submit | --> | review | --> | approve | --> | __end__ | +-----------+ +--------+ +--------+ +----------+ +--------+ | ^ | | | +----------+ | | +----> +--------+ | | revise | ----+ +--------+ =============================================================================== 🚀 开始贷款审批流程... 📄 [张三] 提交申请，收入=4万，信用分=70 🔍 第 1 次审核 张三... ✏️ 张三 正在修改材料：请补充近6个月银行流水（当前收入低于5万） 📄 [张三] 提交申请，收入=7万，信用分=70 🔍 第 2 次审核 张三... ✅ 贷款已批准！恭喜 张三 📌 最终状态: {'applicant': '张三', 'income': 7, 'credit_score': 70, 'missing_docs': '', 'attempts': 2, 'status': 'approved'}💡 这个示例如何体现“返回上一层”？业务动作LangGraph 实现审核员打回申请review 节点的条件边返回 "revise"申请人修改材料执行 revise 节点，修正 income 和 missing_docs重新提交审核add_edge("revise", "review") —— 这就是物理意义上的 “返回上一层（review）节点”防止死循环通过 attempts 计数，达到 3 次强制 reject，打断循环
示例 1：基础顺序链（Linear Chain）像工厂流水线，A -> B -> C 一步步往下走。📊 Mermaid 流程图graph TD; START([开始]) --> clean[清洗数据]; clean --> format[格式化输出]; format --> END([结束]);🖥️ 运行生成 ASCII 图from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): text: str def clean(state): return {"text": state["text"].strip()} def format(state): return {"text": f"Result: {state['text']}"} builder = StateGraph(State) builder.add_node("clean", clean).add_node("format", format) builder.add_edge(START, "clean").add_edge("clean", "format").add_edge("format", END) graph = builder.compile() # 打印控制台流程图 print(graph.get_graph().draw_ascii()) # 输出： # +-----------+ +----------+ +------+ # | __start__ | --> | clean | --> | format | --> | __end__ | # +-----------+ +----------+ +------+ # 生成 Mermaid 代码（可复制到支持 Mermaid 的工具查看） print(graph.get_graph().draw_mermaid())示例 2：条件路由（Conditional Branching）类似“智能分拣”，根据业务规则（如关键词、分数）走不同的分支。📊 Mermaid 流程图graph TD; START([开始]) --> classify[分类器]; classify -->|"类别: 数学"| math_node[数学处理器]; classify -->|"类别: 闲聊"| chat_node[闲聊回复]; math_node --> END([结束]); chat_node --> END([结束]);🖥️ 完整带图代码from typing import Literal, TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): query: str category: str def classify(state): cat = "math" if "计算" in state["query"] else "chat" return {"category": cat} def do_math(state): return {"response": "数学答案: 42"} def do_chat(state): return {"response": "闲聊: 今天天气不错"} def route_condition(state) -> Literal["math_node", "chat_node"]: return "math_node" if state["category"] == "math" else "chat_node" builder = StateGraph(State) builder.add_node("classify", classify) builder.add_node("math_node", do_math) builder.add_node("chat_node", do_chat) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route_condition) # 条件分支 builder.add_edge("math_node", END) builder.add_edge("chat_node", END) graph = builder.compile() # 打印出带有分支箭头的 ASCII 图 print(graph.get_graph().draw_ascii()) # 输出会清晰显示 classify 分叉到两个节点示例 3：ReAct 代理循环（Agent Loop）最经典的图：Agent 思考 -> 调用工具 -> 观察结果 -> 再思考... 直到结束。图中包含自循环。📊 Mermaid 流程图graph TD; START([开始]) --> agent[Agent (LLM)]; agent -->|"有工具调用"| tools[执行工具]; agent -->|"无工具调用"| END([结束]); tools --> agent; style tools fill:#f9f,stroke:#333; style agent fill:#bbf,stroke:#333;🖥️ 完整带图代码（使用内置工具条件）from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun # 模拟工具 search = DuckDuckGoSearchRun() tools = [search] model = ChatOpenAI(model="gpt-4o").bind_tools(tools) def call_model(state: MessagesState): return {"messages": [model.invoke(state["messages"])]} builder = StateGraph(MessagesState) builder.add_node("agent", call_model) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", tools_condition) # 内置判断：去 tools 或结束 builder.add_edge("tools", "agent") # 关键：从 tools 回到 agent（形成循环） graph = builder.compile() # 打印循环流程图 print(graph.get_graph().draw_ascii()) # 输出： # +-----------+ +-------+ +-------+ # | __start__ | --> | agent | --> | tools | # +-----------+ +-------+ +-------+ # ^ | | # | +-----------+ # | (agent 无工具调用时直接结束)示例 4：并行分发（Map-Reduce / Send API）一个节点分发多个并行任务，所有任务完成后自动汇总。📊 Mermaid 流程图graph TD; START([开始]) --> splitter[任务分发器]; splitter -->|"Send(worker, url1)"| worker1[Worker 1]; splitter -->|"Send(worker, url2)"| worker2[Worker 2]; splitter -->|"Send(worker, url3)"| worker3[Worker 3]; worker1 --> aggregator[汇总器]; worker2 --> aggregator; worker3 --> aggregator; aggregator --> END([结束]);🖥️ 完整带图代码import operator from typing import Annotated, List, TypedDict from langgraph.graph import StateGraph, START, END, Send class OverallState(TypedDict): urls: List[str] collected: Annotated[list, operator.add] def splitter(state): # 动态分发并行任务 return [Send("worker", {"url": url}) for url in state["urls"]] def worker(state): return {"collected": [f"抓取结果: {state['url']}"]} def aggregator(state): return {"final": f"总数据量: {len(state['collected'])}"} builder = StateGraph(OverallState) builder.add_node("splitter", splitter) builder.add_node("worker", worker) builder.add_node("aggregator", aggregator) builder.add_edge(START, "splitter") builder.add_conditional_edges("splitter", lambda s: s, path_map=["worker"]) # 关键：分发 builder.add_edge("worker", "aggregator") builder.add_edge("aggregator", END) graph = builder.compile() print(graph.get_graph().draw_ascii()) # 图中会显示 splitter 有多条线指向 worker（并行感）🚀 进阶：生成高清 PNG 图片（可视化终极方案）如果你想直接生成一张 漂亮的 PNG 流程图 用于文档或 PPT，运行下面代码（需要安装 pygraphviz）：pip install pygraphviz# 生成并保存为高清图片 png_data = graph.get_graph().draw_mermaid_png() with open("my_langgraph_flow.png", "wb") as f: f.write(png_data) print("流程图已保存为 my_langgraph_flow.png")💡 针对你的需求总结以上所有示例，你只需要复制代码运行，就能在终端看到 ASCII 字符画流程图。如果是用 Jupyter Notebook，graph.get_graph().draw_mermaid() 会直接渲染出漂亮的矢量图。
项目结构llamaindex-rag-example/├── data/ # 存放待处理的原始文档（如PDF/TXT/Markdown等）├── storage_exp/ # 持久化存储向量索引的目录（自动生成）├── 01_build.py # 构建并持久化向量索引的脚本└── 02_query.py # 加载索引并进行问答查询的脚本PDF文件转Markdown#如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整 官网：https://mineru.net/ 01_build.py 构建索引import os import time import torch from pathlib import Path from llama_index.core import VectorStoreIndex, Settings, Document from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core.node_parser import SimpleNodeParser # 配置 device = "cpu" #如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整https://mineru.net/ markdown_path = "./data/test_manual.md" persist_dir = "./storage_exp" # 创建目录 os.makedirs(persist_dir, exist_ok=True) os.makedirs("./models", exist_ok=True) # 2. 配置全局 Embedding（本地 Qwen3 模型） EMBEDDING_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" Settings.embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_PATH, trust_remote_code=True ) # 读取 Markdown 文件 print(f"读取 Markdown: {markdown_path}") with open(markdown_path, "r", encoding="utf-8") as f: markdown_content = f.read() print(f"文档长度: {len(markdown_content)} 字符") # 创建 Document doc = Document( text=markdown_content, metadata={ "source": Path(markdown_path).name, "file_path": str(Path(markdown_path).absolute()), } ) # 切分文档 print("切分文档...") parser = SimpleNodeParser(chunk_size=512, chunk_overlap=50) nodes = parser.get_nodes_from_documents([doc]) print(f"生成 {len(nodes)} 个节点") # 构建索引 print("构建索引...") start_time = time.time() index = VectorStoreIndex(nodes) build_time = time.time() - start_time print(f"索引构建完成，耗时 {build_time:.2f} 秒") # 保存索引 print(f"保存索引到 {persist_dir}...") index.storage_context.persist(persist_dir=persist_dir) # 统计 index_size = 0 for root, dirs, files in os.walk(persist_dir): for file in files: index_size += os.path.getsize(os.path.join(root, file)) print("\n" + "="*50) print("构建完成!") print("="*50) print(f"文档长度: {len(markdown_content):,} 字符") print(f"节点数量: {len(nodes)}") print(f"索引大小: {index_size / 1024:.1f} KB") print(f"总耗时: {build_time:.2f} 秒") print("="*50) 02_query.pyimport os import sys import time from llama_index.core import StorageContext, load_index_from_storage, Settings from llama_index.llms.openai import OpenAI from llama_index.llms.openai_like import OpenAILike from llama_index.core.prompts import PromptTemplate from llama_index.core.postprocessor import SentenceTransformerRerank from llama_index.retrievers.bm25 import BM25Retriever # 导入依赖 from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings from llama_index.embeddings.huggingface import HuggingFaceEmbedding import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() QIANWEN_KEY = os.getenv("QIANWEN_API_KEY") os.environ["TOKENIZERS_PARALLELISM"] = "false" # 配置 device = "mps" persist_dir = "./storage_exp" lm_studio_url = "http://localhost:1234/v1" # 2. 配置全局 Embedding（本地 Qwen3 模型） EMBEDDING_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" Settings.embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_PATH, trust_remote_code=True ) # 1. 配置全局 LLM（通义千问） Settings.llm = OpenAILike( model="qwen3.5-flash", api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", api_key=QIANWEN_KEY, is_chat_model=True, temperature=0.1 ) # 重排序器 # 中文专用重排模型 # reranker = SentenceTransformerRerank( # model="BAAI/bge-reranker-base", # top_n=5 # ) # 提示词模板 PROMPT = PromptTemplate( """你是一个专业的 HR 助手，请基于以下员工手册内容回答问题。 {context_str} 问题：{query_str} 回答：""" ) # 加载索引 print("加载索引...") storage_context = StorageContext.from_defaults(persist_dir=persist_dir) index = load_index_from_storage(storage_context) print(f"索引加载成功，共 {len(index.docstore.docs)} 个节点") def hybrid_search(query, top_k=5): """混合检索""" # 向量检索 vector_retriever = index.as_retriever(similarity_top_k=10) vector_nodes = vector_retriever.retrieve(query) # BM25 检索 all_nodes = [] for doc_id in index.docstore.docs.keys(): doc = index.docstore.get_document(doc_id) if hasattr(doc, 'nodes'): all_nodes.extend(doc.nodes) else: all_nodes.append(doc) bm25_retriever = BM25Retriever.from_defaults( nodes=all_nodes, similarity_top_k=10, ) bm25_nodes = bm25_retriever.retrieve(query) # 合并去重 seen = set() merged = [] for node in vector_nodes + bm25_nodes: if node.node.node_id not in seen: seen.add(node.node.node_id) merged.append(node) return merged[:top_k] # # 重排序 # reranked = reranker.postprocess_nodes(merged, query_str=query) # return reranked[:top_k] def generate_answer(query, nodes): """生成回答""" from llama_index.core import get_response_synthesizer synthesizer = get_response_synthesizer(text_qa_template=PROMPT) return synthesizer.synthesize(query, nodes=nodes) def query(question): """执行查询""" nodes = hybrid_search(question) response = generate_answer(question, nodes) return response.response if __name__ == "__main__": question = "员工违约的条件有哪些？" print(f"\n问题: {question}") print("-" * 40) start = time.time() answer = query(question) elapsed = time.time() - start print(f"回答: {answer}") print("-" * 40) print(f"耗时: {elapsed:.2f} 秒") 测试结果/opt/miniconda3/envs/langchain-dev/bin/python /Users/t-mac/workspace/Python/python_code/24-LlamaIndex/projects/llamaindex-rag-example/02_query.py W0522 23:40:21.752000 33319 site-packages/torch/distributed/elastic/multiprocessing/redirects.py:35] NOTE: Redirects are currently not supported in MacOs. W0522 23:40:21.774000 33319 site-packages/torch/utils/_pytree.py:630] <enum 'KernelPreference'> is an Enum subclass and is now natively supported by torch.compile as an opaque value type. Calling register_constant() on Enum subclasses is deprecated and will be an error in a future release. W0522 23:40:21.793000 33319 site-packages/torch/utils/_pytree.py:630] <enum 'ScaleCalculationMode'> is an Enum subclass and is now natively supported by torch.compile as an opaque value type. Calling register_constant() on Enum subclasses is deprecated and will be an error in a future release. Loading weights: 100%|██████████| 310/310 [00:00<00:00, 8048.99it/s] 加载索引... 索引加载成功，共 98 个节点 问题: 员工违约的条件有哪些？ ---------------------------------------- 回答: 你好！我是你的专业 HR 助手。根据提供的《员工手册》内容，虽然手册中主要使用“违纪”、“严重违纪”及“解除劳动合同”等术语，但以下情况均属于员工违反公司规章制度或劳动合同约定的条件（即通常理解的“违约”情形）。 以下是基于文档整理的员工违纪及违规的具体条件分类： ### 1. 考勤与出勤违规 * **迟到早退：** 连续 30 天内迟到或早退三次及以上（10.1.1）；试用期内迟到早退累计达三次或以上（4.5）。 * **旷工：** * 未经请假或请假未获批准擅自不到岗（3.1）。 * 请假期满未续假或续假未批逾期不归（3.3.2）。 * 特殊原因未在当日上班 1 小时内请假，或事后 3 个工作日内未补办手续（3.3.3）。 * 不服从工作调动、工时制度调整拒不执行导致未正常出勤（3.3.4）。 * **后果阈值：** 连续旷工 2 天及以上，或一年内累计旷工 3 天及以上（10.1.2）；连续旷工 3 天及以上，或全年累计旷工 7 天及以上（3.2）。 * **打卡异常：** 委托他人打卡或替他人打卡（10.1.3）；连续 30 天累计 6 次以上未打卡且无法说明原因或提供证明（1.4 / 10.1.4）。 * **试用期出勤率：** 试用期实际出勤日/应出勤日低于 85%（4.5）。 ### 2. 诚信与入职信息违规 * **虚假信息：** 伪造学历、工作经历、证书，简历、入职登记信息、健康状况等与真实情况不符（4.6）。 * **隐瞒欺骗：** 有欺骗、隐瞒行为（4.6）。 * **价值观问题：** 诚信等价值观存在问题，无论服务年限长短，均可能导致立即解除劳动合同（1.3）。 * **竞业限制：** 入职本公司将违反竞业限制义务（4.15）。 ### 3. 行为规范与安全违规 * **扰乱秩序：** 发起、参与离岗、怠工、罢工、示威，或鼓动他人参与（10.2.1）。 * **骚扰行为：** 侮辱、诽谤、辱骂、造谣、威胁、恐吓他人（10.2.2）。 * **现场管理：** 将无关人员带入工作现场，情节严重或造成损害（10.2.3）。 * **安全隐患：** 在库房等高风险区域内吸烟，情节严重或造成损害（10.2.4）。 * **资产损害：** 损坏或遗失公司财物，浪费公司资产，情节严重或造成损害（10.2.5 / 10.2.6）。 ### 4. 法律、健康与个人资质违规 * **刑事犯罪：** 被判处有期徒刑尚在服刑（4.8）；通缉在案（4.10）；实施违反国家法律法规的行为（如加入邪教、恐怖组织、酗酒、吸毒、赌博、嫖娼等）（4.14）。 * **权利状态：** 被剥夺公民权利者（4.9）。 * **年龄与健康：** 未满 16 周岁（4.11）；传染性疾病病原携带者或患有其他严重疾病（4.12）；患有精神病（4.13）。 * **债务问题：** 拖欠他人重大债务或进入法院执行程序未偿还债务者（4.16）。 * **过往记录：** 曾经被公司辞退或未经批准擅自离职者（4.7）。 ### 5. 试用期特殊解除条件 * **处分记录：** 试用期期间受记过及以上处分者（4）。 * **综合表现：** 上述考勤、诚信、法律等条款在试用期内触犯者（4.5 - 4.16）。 ### 6. 管理与连带责任 * **监管不力：** 管理者对员工违纪行为存在监管不力、失察、包庇、纵容或教唆等情形的，主管也将受到处罚（9.6）。 * **兜底条款：** 其他任何与前述违纪行为严重性相当的行为，或违反部门规定的行为，情节较重或造成一般损害的（5）。 **总结：** 根据手册第 10 条，若员工出现上述**严重违纪行为**，公司将对其进行**辞退（解除劳动合同）**。其中涉及诚信价值观问题、刑事犯罪、严重旷工等行为，公司有权立即解除合同且不支付经济补偿。 希望以上整理能帮助您准确理解员工手册中的相关规定。如有具体案例需要分析，欢迎随时提问。 ---------------------------------------- 耗时: 37.51 秒
LangGraph# 什么是 LangGraph？ # LangGraph 是一个用于构建状态化、多步骤 AI 应用的框架。它使用图（Graph） 的概念来组织工作流： # 节点（Nodes）：图中的处理单元，可以是 LLM 调用、工具执行或自定义函数 # 边（Edges）：连接节点的路径，定义执行顺序 # 状态（State）：在节点之间传递的数据结构 # LangGraph vs LangChain # 特性 LangChain LangGraph # 抽象级别 高级 低级 # 适用场景 快速构建标准 Agent 复杂自定义工作流 # 控制粒度 通过中间件 完全控制每个节点 # 状态管理 自动 手动但灵活 # Hello World Graph # 1. 定义状态 - 使用 TypedDict from typing import TypedDict, Annotated,Dict from langgraph.graph import StateGraph class AgentState(TypedDict): message: str def greeting_node(state: AgentState)-> AgentState: """ 一个向状态添加消息的简单节点 """ state['message'] = "Hello" + state["message"] return state graph = StateGraph(AgentState) graph.add_node("greeter",greeting_node) graph.set_entry_point("greeter") graph.set_finish_point("greeter") app = graph.compile() 打印图流程from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) 多个节点并添加边#多个节点并添加边 from typing import TypedDict, Annotated,Dict from langgraph.graph import StateGraph class AgentState(TypedDict): name :str age : str final:str def first_node(state: AgentState)-> AgentState: """这是第一个节点""" state["final"] = f"你好 {state['name']}" return state def second_node(state: AgentState)-> AgentState: """这是第二个节点""" state["final"] = state["final"] + f"你已经 {state['age']}了" return state graph = StateGraph(AgentState) graph.add_node("first_node",first_node) graph.add_node("second_node",second_node) graph.set_entry_point("first_node") graph.add_edge("first_node","second_node") graph.set_finish_point("second_node") app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) result = app.invoke({"name":"zhangsan","age":"18",}) print(result) 加模型顺序执行#加模型顺序执行 from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from typing import TypedDict, Annotated,Dict # ========================================== # 1. 初始化模型 # ========================================== model = ChatOllama( model="qwen3.5:2b", base_url="http://localhost:11434", temperature=0.7, ) class SimpleState(TypedDict): input_text:str processed_text :str llm_response :str final_output:str # 定义节点函数 def preprocess(state: SimpleState) -> dict: """预处理节点：清理和格式化输入""" text = state["input_text"].strip().lower() print(f" [预处理] 输入: '{state['input_text']}' -> '{text}'") return {"processed_text": text} def call_llm(state: SimpleState) -> dict: """LLM 节点：调用语言模型""" messages = [ {"role": "system", "content": "你是一个友好的助手，请简洁回答问题。"}, {"role": "user", "content": state["processed_text"]} ] response = model.invoke(messages) print(f" [LLM] 响应: {response.content[:50]}...") return {"llm_response": response.content} def postprocess(state: SimpleState) -> dict: """后处理节点：格式化输出""" final = f"✨ AI 回复：{state['llm_response']}" print(" [后处理] 完成格式化") return {"final_output": final} # 构建图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("preprocess", preprocess) graph.add_node("call_llm", call_llm) graph.add_node("postprocess", postprocess) # 添加边（定义执行顺序） graph.add_edge(START, "preprocess") graph.add_edge("preprocess", "call_llm") graph.add_edge("call_llm", "postprocess") graph.add_edge("postprocess", END) # 编译图 app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) result = app.invoke({"input_text": " 介绍一下你自己 "}) print(f"\n最终输出：\n{result['final_output']}") 条件分支工作流from typing import Literal #条件分支工作流 #加模型顺序执行 from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from typing import TypedDict, Annotated,Dict class AgentState(TypedDict): query: str query_type: str response: str def classify_query(state: AgentState) ->dict: """分类节点：判断查询类型""" query = state["query"].lower() if any(word in query for word in ["天气", "温度", "下雨"]): query_type = "weather" elif any(word in query for word in ["计算", "加", "减", "乘", "除", "等于"]): query_type = "math" else: query_type = "general" print(f" [分类] 查询类型: {query_type}") return {"query_type": query_type} def handle_weather(state: AgentState) -> dict: """处理天气查询""" print(" [天气处理] 执行天气查询逻辑...") # 实际应用中这里会调用天气 API response = "🌤️ 今天天气晴朗，温度 25°C，适合外出！" return {"response": response} def handle_math(state: AgentState) -> dict: """处理数学计算""" print(" [数学处理] 执行计算逻辑...") content = "计算器损坏，计算结果统一为 0。" return {"response": f"🔢 {content}"} def handle_general(state: AgentState) -> dict: """处理一般查询""" print(" [通用处理] 执行通用 LLM 调用...") messages = [ {"role": "system", "content": "你是一个友好的助手，请简洁回答问题。"}, {"role": "user", "content": state["query"]} ] result = model.invoke(messages) return {"response": f"💡 {result.content}"} def route_query(state: AgentState) -> Literal["weather", "math", "general"]: """路由函数：返回下一个节点名称""" return state["query_type"] # 构建图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("classify", classify_query) graph.add_node("weather", handle_weather) graph.add_node("math", handle_math) graph.add_node("general", handle_general) # 添加边 graph.add_edge(START, "classify") # 添加条件边 graph.add_conditional_edges( "classify", # 从哪个节点出发 route_query, # 路由函数 { # 路由映射 "weather": "weather", "math": "math", "general": "general" } ) # 所有处理节点都到 END graph.add_edge("weather", END) graph.add_edge("math", END) graph.add_edge("general", END) # 编译 app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) # 测试不同类型的查询 test_queries = [ "今天北京的天气怎么样？", "计算 123 加 456 等于多少？", "Python 是什么编程语言？" ] for query in test_queries: print(f"\n查询: {query}") result = app.invoke({"query": query}) print(f"响应: {result['response'][:100]}...") 带内存的对话工作流from typing import Literal from langgraph.graph.message import add_messages # 必须导入！ from langchain_core.messages import BaseMessage from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated,Dict from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.checkpoint.memory import MemorySaver # 带内存的对话工作流 # 使用 add_messages 注解，消息会自动追加 class AgentState(TypedDict): messages: Annotated[list, add_messages] turn_count: int # model = ChatOllama( # model="qwen3.5:9b", # base_url="http://localhost:11434", # temperature=0.7, # # M5 专属优化 # num_ctx=1024, # 越小越快 # num_predict=256, # 限制回答长度 # use_metal=True, # ✅ M5 必开：调用GPU/NPU # # keep_alive="24h", # 模型常驻，不卸载 # ) model = ChatOpenAI( base_url="http://localhost:1234/v1", api_key="dummy-key", # 随便填 model="qwen/qwen3.5-9b", temperature=0.2, max_tokens=1024, # max_tokens=64, # ✅ 核心优化 2：关闭思考链（关键！） # 关键 2：强制关闭思考链（这个格式 100% 生效） # use_metal=True, # streaming=True, # 开启流式输出 # callbacks=[StreamingStdOutCallbackHandler()], # 实时打印 ) def chat_node(state: AgentState) -> dict: """对话节点：处理用户消息并生成回复""" # 添加系统提示（如果是第一轮） messages = state["messages"] if not any(isinstance(m, SystemMessage) for m in messages): messages = [ SystemMessage(content="你是友好助手，请直接回答问题，**绝对不要输出思考过程，不要写推理步骤**") ] + messages # 调用模型 response = model.invoke(messages) # 更新轮数 turn_count = state.get("turn_count", 0) + 1 print(f" [对话轮次 {turn_count}] AI: {response.content[:50]}...") # 返回新消息（会自动追加到 messages 列表） return { "messages": [response], # add_messages 会自动追加 "turn_count": turn_count } def should_continue(state: AgentState) -> Literal["continue", "end"]: """决定是否继续对话""" # 这里简化为总是返回 end，实际应用中可以检查用户意图 return "end" # 构建图 graph = StateGraph(AgentState) graph.add_node("chat", chat_node) graph.add_edge(START, "chat") graph.add_edge("chat", END) # 简化：每次调用处理一轮 # 使用内存保存器 memory = MemorySaver() app = graph.compile(checkpointer=memory) # 模拟多轮对话（使用相同的 thread_id） config = {"configurable": {"thread_id": "user_001"}} conversations = [ "你好！我叫小明。", "我最喜欢的编程语言是 Python。", "你还记得我的名字吗？", "我喜欢什么编程语言？" ] for user_input in conversations: print(f"\n用户: {user_input}") result = app.invoke( {"messages": [HumanMessage(content=user_input)]}, config=config ) print(f"AI: {result['messages'][-1].content}")
数据向量化#读取 现有json文件，不需要分割的数据直接构建 import json from langchain_experimental.text_splitter import SemanticChunker from langchain_community.vectorstores import FAISS from langchain_core.documents import Document # 读取文件内容 with open("./iciba_data_multi.json", encoding="utf-8") as f: json_data = json.load(f) # 加载embedding模型 from langchain_huggingface import HuggingFaceEmbeddings MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" # mps mac芯片mps加速 embeddings = HuggingFaceEmbeddings( model_name=MODEL_PATH, model_kwargs={"device": "mps"}, encode_kwargs={"normalize_embeddings": True} ) # 2. 准备文档列表 # 这里我们不再使用 text_splitter，而是手动构建 Document 列表 documents = [] for item in json_data: # 拼接文本：英文+中文 full_text = f"{item['content']} {item['note']}" # 构建 LangChain 的 Document 对象 # page_content 是要向量化的内容 # metadata 是你想保留但不参与向量化的信息（比如日期） doc = Document( page_content=full_text, metadata={"date": item["date"]} ) documents.append(doc) # 从文档列表创建 FAISS 向量库 vector_store = FAISS.from_documents(documents=documents, embedding=embeddings) # 【可选】持久化存储：将索引保存到本地磁盘 vector_store.save_local("faiss_index") Ollama 运行 Qwen3.5-2B❯ ollama run qwen3.5:2b pulling manifest pulling b709d81508a0: 100% ▕██████████████████████████████████████▏ 2.7 GB pulling 9be69ef46306: 100% ▕██████████████████████████████████████▏ 11 KB pulling 9371364b27a5: 100% ▕██████████████████████████████████████▏ 65 B pulling ee043a99abe5: 100% ▕██████████████████████████████████████▏ 473 B verifying sha256 digest writing manifest success >>> 线阶段：RAG 问答# 在线阶段：RAG 问答 # 1. 核心思路 # 创建一个 “检索工具”，AI 回答问题前，先调用这个工具找相关文档，再结合文档回答（避免瞎编）。 import json from langchain.agents import create_agent from langchain_core.tools import tool from langchain_ollama import ChatOllama # ========================================== # 1. 初始化模型 # ========================================== model = ChatOllama( model="qwen3.5:2b", base_url="http://localhost:11434", temperature=0.7, ) # ========================================== # 2. 模拟你的知识库数据 (实际使用时请替换为你的向量库检索逻辑) # ========================================== # 这里为了演示，我直接把你的 JSON 数据放在这里，实际应替换为 vector_store.similarity_search(query) @tool def search_docs(query: str) -> str: """ 在文档库中搜索相关信息。 如果用户询问关于日期、格言或特定事件的内容，请使用此工具。 """ # --- 实际逻辑替换结束 --- formatted_results = [] results = ensemble_retriever.invoke(query) for doc in results: text = f"内容: {doc.page_content}" formatted_results.append(text) return "\n---\n".join(formatted_results) # ========================================== # 3. 创建 AI Agent # ========================================== agent = create_agent( model=model, tools=[search_docs], system_prompt="""你是每日一词的助手。 1. 回答用户问题前，必须先调用 search_docs 工具查找信息。 2. 只能根据搜索到的 [内容] 和 [备注] 进行回答。 3. 如果搜索不到相关信息，请直接说“我在日记中没找到相关内容”，不要编造。 4. 回答时直接输出工具查询的内容。 """ ) # ========================================== # 4. 问答循环 # ========================================== print("🤖 助手已就绪，请问关于这些日期的格言，你想知道什么？") while True: try: query = input("\n请输入你的问题 (输入 q 退出): ") if query.lower() == 'q': break # 调用 Agent # 注意：这里使用 invoke 传入字典 response = agent.invoke({ "messages": [{"role": "user", "content": query}] }) # 【关键优化】：正确提取回复内容 # response["messages"] 是一个列表，[-1] 是最后一条消息对象 # .content 是获取该对象中的文本内容 final_answer = response["messages"][-1].content print("\n📝 助手回答：") print(final_answer) print("-" * 30) except Exception as e: print(f"发生错误: {e}") 结果
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]
# 核心概念 # 验证和重试 = 确保 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()