【LLM相关知识点】关于LangChain框架学习简单整理（三）

【LLM相关知识点】关于LangChain框架学习简单整理（三）

一、核心模块和协作模式

参考极简LangChain智能体开发入门指南，LangChain官方文档

LangChain核心模块与功能：

核心模块	功能描述	关键技术点
​模型I/O	管理大模型输入输出，包含提示模板、模型调用接口、输出解析器	支持动态变量注入的提示模板（如PromptTemplate） ；输出结构化解析（如JSON/Pydantic）
​数据连接	实现外部数据加载、向量化存储与检索	文档分块（TextSplitters）、向量存储（Faiss/Pinecone） ；支持PDF/网页等多种数据源
​链（Chains）​	串联多个处理步骤形成端到端任务流	简单链（LLMChain）、复杂链（SequentialChain/RouterChain） ；支持异步/流式处理
​记忆（Memory）​	维护对话历史或任务上下文	短期记忆（ConversationBufferMemory）、长期记忆（RedisMemory） ；支持状态持久化
​代理（Agents）​	动态调用外部工具（API/数据库）	ReAct代理实现推理-行动循环 ；工具集成（如天气API/搜索工具）
​回调（Callbacks）​	监控任务执行过程，处理日志/异常	支持日志记录、性能分析及自定义事件触发

模块协作模式（以典型场景为例）：

应用场景	模块协作流程	技术实现
​RAG系统	数据连接加载文档→分块→向量化存储 → 链模块组合检索与生成 → 代理补充实时数据 → 记忆保存历史	使用RetrievalQAChain组合检索器与LLM；代理调用知识库工具
​智能客服	代理调用知识库工具 → 记忆模块维护多轮对话 → 输出解析器提取结构化数据	ConversationChain管理对话流；OutputParser生成工单
​自动化测试	代理生成测试用例 → 调用Selenium执行 → 回调模块记录结果	自定义工具集成Selenium；回调函数记录执行日志

二、LangChain的5种Agentic设计模式

参考 LangChain中ReAct的实现原理浅析，一文读懂AI Agent原理及开发方法，基于Langchain的Agent实战，实现Agent对话

​模式	​优势	​适用场景
反思模式	自我优化能力，​动态适应性	单任务质量敏感型场景（如法律文档生成）
工具模式	扩展性高，支持任意外部工具	API集成、数据查询类任务
ReAct模式 / 零样本React	简单快速部署	通用问答、快速原型开发
规划模式	任务分解清晰	项目管理系统、多阶段流程
多智能体模式	任务分解与协作，异构能力整合，容错性与效率	跨领域协作型场景（如供应链优化）

参考 5种Agentic设计模式：langchain实现

1、反思模式（Reflection Pattern）：基于两类提示词，实现自我批评

• 在这种模式下，Agent会“审查”自己的工作——发现错误并不断迭代，直到最终结果完美无缺。

• 可以将其想象成一个自我批评的循环，模型会在呈现结果之前完善每个草稿！

这里涉及生成（generate）和Reflect（反思）两个部分：

• generate：

  GENERATION_PROMPT = """您的任务是针对用户的请求生成尽可能最好的内容。如果用户提出批评，请使用您之前尝试的修订版本进行回应。您必须始终输出修订后的内容。""" 
  

• Reflect

  REFLECTION_PROMPT = """您的任务是针对用户生成的内容生成批评和建议。如果用户内容存在错误或需要改进，请输出建议和批评列表。如果用户内容没有问题并且无需更改，请输出以下内容：<OK>"""
  

• langchian实现方案

  from langgraph.graph import MessageGraphbuilder = MessageGraph()
  builder.add_node("generate", generation_node)
  builder.add_node("reflect", reflection_node)
  builder.set_entry_point("generate")
  def should_continue(state: List[BaseMessage]):if len(state) > 6:return ENDreturn"reflect"builder.add_conditional_edges("generate", should_continue)
  builder.add_edge("reflect", "generate")
  graph = builder.compile()ompile()uilder.compile()
  

2、工具模式（Tool Use Pattern）：LLM调用外部Function，融合输出结果

• 将使用Tool的agent看做是一个人，LLM 是它的大脑，一套Tool是它采取行动的手。

• LLM不再仅仅依赖其内部知识，而是利用Tool查询数据库、调用API和执行Python脚本，调用外部知识完成工作。

langChain的实现：

from langchain_core.tools import StructuredTooldef multiply(a: int, b: int) -> int:"""Multiply two numbers."""return a * basync def amultiply(a: int, b: int) -> int:"""Multiply two numbers."""return a * bcalculator = StructuredTool.from_function(func=multiply, coroutine=amultiply)print(calculator.invoke({"a": 2, "b": 3}))
print(await calculator.ainvoke({"a": 2, "b": 5})
)

1）什么是Function Calling：意图识别 → 结构化转换 → 函数调用

参考什么是Function Calling，它是如何增强大模型能力的？今天一文搞懂！

Function Calling是大模型在对话过程中调用外部函数的能力，它允许模型在无法直接回答问题时，返回一个函数调用请求（高度结构化的JSON数据），而不是生成纯文本。开发者需要手动解析这个请求，执行对应的函数，并将结果传回给模型，以生成最终答案。

Function Calling的核心机制就是通过大模型对用户输入的意图进行识别，生成结构化请求（如 JSON Schema），再触发外部函数的调用，流程图如下：

a）核心机制：LLM具备Text2Schema（API / SQL等）

1. ​意图识别与参数生成
   模型基于用户输入的自然语言，识别需要调用的函数并生成结构化参数（通常为JSON格式）。例如，用户提问“巴黎今天多少度？”，模型可能生成：

   {"name": "get_weather", "arguments": {"location": "Paris"}}
   

2. ​函数执行与结果整合
   开发者需手动执行对应的函数或API调用，并将结果返回给模型，模型再生成最终的自然语言回复。例如调用天气API获取温度数据后，模型生成：“巴黎今天气温22°C”。

3. ​标准化接口设计
   函数需预先定义名称、描述、参数格式（类型、必填项、枚举值等），确保模型与外部服务的交互规范性。例如定义天气查询函数时需明确location为必填字符串参数。

b）如何让大模型支持Function Calling：函数库建立 + 标注函数调用的信息 + 调用决策模块

要让大模型支持Function Calling，通常需要以下几个关键步骤：用户输入文本 + 函数库建立 + 标注函数调用的信息 -> LLM输出结果 -> 调用决策模块，判断是否调用函数；

模型架构与设计：

• 增加相关模块：设计专门的模块来处理函数调用相关的逻辑，例如，添加一个 「调用决策模块」，用于判断输入是否需要调用函数以及调用哪个函数；同时设置一个 「结果融合模块」，将函数调用的结果与模型的其他输出进行合理融合。

• 参数调整：通过大量训练数据，调整模型参数，使模型能够准确的理解何时需要调用函数，以及根据函数的输出进行后续处理。

训练数据准备：

• 标注函数调用信息：标注函数调用信息，比如，如果输入是 「查询 2024 年北京的人口数量」，则可以标注出需要调用一个查询人口数据的函数，函数参数为 「2024 年」 和 「北京」。

• 多样化样例：构建多样化的数据集，涵盖各种不同类型的函数调用场景。

函数库集成：

• 建立函数库：创建一个包含各种可调用函数的函数库，函数库中的函数可以是实现特定功能的代码模块，如数据查询、数学计算、文本处理等。每个函数都有明确的输入参数和输出格式。

• 接口设计：为函数库设计统一的接口，以便模型能够方便地调用函数。接口要能够接收模型传递的参数，并将函数的输出返回给模型。同时，要确保接口的稳定性和兼容性，以保证函数调用的顺利进行。

推理执行：

• 调用决策：在推理阶段，当模型接收到输入时，首先由调用决策模块根据模型的输出和相关规则，判断是否需要调用函数。如果需要，确定要调用的函数及其参数。

• 函数调用与结果处理：通过接口调用函数库中的相应函数，并将函数的输出传递给结果融合模块。结果融合模块将函数调用的结果与模型的其他输出进行整合，生成最终的输出结果。例如，如果函数调用的结果是一个具体的数据，而模型原本的输出是一段文本，那么结果融合模块可能会将数据嵌入到文本中合适的位置，形成一个完整的回答。

c）和Tool / Plugins的区别：Tool / Plugins更多是指令标识，并不包含意图识别过程种对Schema的转换

参考 Function Calling在大模型中的崛起：原理与应用场景解析

在对话式AI中，Tools/Plugins通常指模型可以通过特殊的提示或访问外部资源（如web search plugin、文档检索、数据库查询等），来完成用户请求，相比之下：

Tools/Plugins

• 多用“自然语言”或“命令式提示”让模型访问外部资源；

• 可能在返回结果前就需要模型自行理解并转换。

Function Calling

• 更注重结构化参数 的传递与函数调用，让模型可以精准地给出调用意图及参数，而不是模糊地发起plugin请求；

• 更适合需要定制函数或严谨参数场景，如财务计算、CRM查询、客户信息编辑等。

• Tools/Plugins： 常见于搜索引擎、外部API，对模型结果更泛化且易耦合，但是参数可控性、结果回传往往不如Function Calling严谨。

• Function Calling： 适合企业内需严格JSON schema接口的函数调用，对数据安全、准确度、审计有较高要求，适合“企业内业务逻辑”执行。

特性	Tools/Plugins	Function Calling
调用方式	自然语言+提示词	结构化参数调用（JSON Schema）
适用场景	搜索、外部 API、数据查询	CRM、财务计算、订单管理等内部业务
数据传递	模型自行解析，结果可能不精确	严格按照预定义的参数结构传递数据
可控性	低，模型可能错误调用	高，调用意图可审计、可跟踪
安全性	低，结果依赖 AI 理解能力	高，可对 API 访问进行严格控制
适合企业应用	适合泛化任务，如信息检索	适用于高安全、高可控业务场景

2）Function Calling 和 MCP的区别：MCP是AI应用间的数据通信方式

参考 什么是Function Calling，它是如何增强大模型能力的？今天一文搞懂！

Function Calling是大模型的一种功能（实现流程如上一小节所示）：允许模型根据用户输入识别意图并触发预先设定的外部函数或工具，将自然语言转化为结构化参数传送给函数，再整合结果生成回应。

而MCP是 Anthropic 提出的标准化协议，旨在让 AI 应用（包括 Agent）更好地连接各种外部数据服务，帮助大模型实现更好、更相关的响应，相当于大模型的 「HTTP 协议」。

Function Calling有两个主要用例：

• 获取数据：检索最新信息以合并到模型的响应（RAG）中。用于搜索知识库和从API中检索特定数据（例如当前天气数据）。

• 执行动作：执行提交表单、调用API、修改应用程序状态（UI/前端或后端）或执行代理工作流操作（如移交对话）等操作。

3、推理行动模式（Reasoning and Acting - ReAct）：T-A-O反馈循环模式

• ReAct结合了Reflection和Tool两种模式

• Agent利用Reflection通过 LLM 生成的输出，以及使用Tool与外部世界交互

关于ReAct的prompt：

REACT_PROMPT = """您是一个函数调用 AI 模型。您通过运行循环来操作，步骤如下：思考、行动、观察。在 <tools></tools> XML 标记中为您提供函数签名。您可以调用一个或多个函数来协助用户查询。不要假设要将哪些值插入函数。要特别注意属性“类型”。您应该像在 Python 字典中一样使用这些类型。对于每个函数调用，在 <tool_call></tool_call> XML 标记中返回一个带有函数名称和参数的 json 对象，如下所示：<tool_call>```{"name": <function-name>,"arguments": <args-dict>, "id": <monotonically-increasing-id>}```</tool_call>以下是可用的工具/操作：```<tools>``%s``</tools>````示例会话：`` ``<question>马德里现在的温度是多少？</question>``<thought>我需要了解马德里当前的天气</thought>``<tool_call>{"name": "get_current_weather","arguments": {"location": "Madrid", "unit": "celsius"}, "id": 0></tool_call>`` ``您将再次被调用：`` ``<observation>{0: {"temperature": 25, "unit": "celsius"}></observation>`` ``然后您输出：`` ``<response>马德里当前温度为 25 摄氏度</response>`` ``其他限制：`` ``- 如果用户询问您与上述任何工具无关的问题，请自由回答并使用 <response></response> 标签将您的答案括起来。"""

langChain实现：

from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.tools.tavily_search import TavilySearchResultstools = [TavilySearchResults(max_results=1)]
llm = OpenAI()
agent = create_react_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor.invoke({"input": "what is LangChain?"})

1）ReAct - Few Shot 和 Zero shot

a）ReAct模式的核心机制： T - A - O

ReAct（Reasoning + Acting）是一种结合推理和行动的框架，通过多步骤交互完成任务：

• ​工作流程：

  • ​推理（Reasoning）​：LLM分析问题，拆解步骤（如“需要先查天气，再计算行程时间”）。格式如下：

    Thought: 需要先查询用户账户信息  
    Action: account_query_tool  
    Action Input: {"user_id": "U12345"}
    

  • ​行动（Acting）​：调用工具（Tool）执行操作（如调用天气API）。

  • ​观察（Observation）​：获取工具返回结果，进入下一轮推理。

• ​依赖条件：

  • 通常需要提供少量示例（Few-Shot）在提示词中，指导模型如何拆分步骤和调用工具。

• ​示例：

  # 提示词中可能包含类似示例：
  """
  问题：今天北京的温度是多少？如果高于25度，推荐一个适合的景点。
  思考：首先需要查询天气，然后根据温度结果调用景点推荐工具。
  行动：调用天气查询工具，参数：北京
  观察：北京当前温度28度
  思考：温度高于25度，需要推荐清凉的景点。
  行动：调用景点推荐工具，参数：清凉
  最终答案：建议游览颐和园。
  """
  

b）零样本ReAct（Zero-Shot ReAct）的特点

零样本ReAct是ReAct模式的一种特例，其核心区别在于：

• ​无需示例（Zero-Shot）​：

  • 不依赖任务相关的示例，直接依赖LLM的预训练知识理解任务逻辑。
  • 通过结构化提示模板明确工具描述和流程规则，而非具体案例。

• ​提示词设计：

  # 零样本ReAct的典型提示模板：
  """
  你是一个能够调用工具的助手。以下是可用工具：
  - 天气查询工具：输入城市名，返回温度。
  - 景点推荐工具：输入关键词，返回景点列表。请按以下步骤回答：
  1. 分析问题需要哪些工具。
  2. 按顺序调用工具。
  3. 根据结果生成最终答案。当前问题：{用户输入}
  """
  

• ​优点：

  • 灵活性高：无需为每个任务编写示例，适合快速适配新工具或场景。
  • 依赖模型能力：要求LLM本身具有较强的逻辑推理和工具理解能力（如GPT-4）

2）ReAct 与 Function Calling 的关系：ReAct是Reflect模式和Tool模式的结合

核心分工：

• ​ReAct 框架：负责多步推理（Reasoning）与工具调用（Acting）的流程控制。其核心是通过模型的思考（Thought）生成动作（Action），调用工具后基于观察（Observation）进入下一轮循环。
• ​Function Calling：是大模型自身的能力​（如 GPT-4），用于从自然语言中提取结构化参数，生成符合工具调用规范的输出（如 JSON 格式的 {"name": "tool_name", "arguments": {...}}）。

协同工作场景：

• ​结构化参数生成：
  当 ReAct 代理需要调用工具时，若底层模型支持 Function Calling，可直接利用此能力生成严格符合工具参数结构的输入，避免自然语言解析错误。 例如：

  from langchain.agents import create_react_agent# 假设工具已通过 @tool 定义
  agent = create_react_agent(llm, tools=[search, calculate], prompt=react_prompt)
  

  若 llm 是支持 Function Calling 的模型（如 ChatOpenAI），Agent会在生成 Action 步骤时自动触发函数调用，输出工具参数。

• ​非结构化模型的替代方案：
  若模型不支持 Function Calling，ReAct代理仍可通过提示词工程​（如强制输出 JSON格式）或后解析逻辑​（如正则表达式）提取参数，但可靠性较低。

4、规划模式（Planning Pattern）：任务分解与执行模式

• 规划是决定按照什么步骤顺序来完成一项任务

• 让LLM将任务分解成更小、容易实现的子目标，并且概述目标，就像项目经理组织复杂的工作流程一样。

5、多智能体模式（Multi-Agent Pattern）：多角色分工协作

• 类似CreawAI框架，每个智能体有自己的角色，任务被划分为由不同角色执行的子任务。

• 例如一个工程团队，PM设定目标，技术主管设计，开发人员编码，DevOps工程师处理部署。他们一起交付统一的产品！

• 基于langchain、LangGraph实现：

  from typing import Annotatedfrom langchain_community.tools.tavily_search import TavilySearchResults
  from langchain_core.tools import tool
  from langchain_experimental.utilities import PythonREPLfrom typing import Literal
  from typing_extensions import TypedDictfrom langchain_anthropic import ChatAnthropic
  from langgraph.graph import MessagesState, END
  from langgraph.types import Commandfrom langchain_core.messages import HumanMessage
  from langgraph.graph import StateGraph, START, END
  from langgraph.prebuilt import create_react_agenttavily_tool = TavilySearchResults(max_results=5)# This executes code locally, which can be unsafe
  repl = PythonREPL()@tool
  def python_repl_tool(code: Annotated[str, "The python code to execute to generate your chart."],):"""Use this to execute python code and do math. If you want to see the output of a value,you should print it out with `print(...)`. This is visible to the user."""try:result = repl.run(code)except BaseException as e:return f"Failed to execute. Error: {repr(e)}"result_str = f"Successfully executed:\n```python\n{code}\n```\nStdout: {result}"return result_strmembers = ["researcher", "coder"]options = members + ["FINISH"]system_prompt = ("You are a supervisor tasked with managing a conversation between the"f" following workers: {members}. Given the following user request,"" respond with the worker to act next. Each worker will perform a"" task and respond with their results and status. When finished,"" respond with FINISH."
  )class Router(TypedDict):"""Worker to route to next. If no workers needed, route to FINISH."""next: Literal[*options]llm = ChatAnthropic(model="claude-3-5-sonnet-latest")class State(MessagesState):next: strdef supervisor_node(state: State) -> Command[Literal[*members, "__end__"]]:messages = [{"role": "system", "content": system_prompt},] + state["messages"]response = llm.with_structured_output(Router).invoke(messages)goto = response["next"]if goto == "FINISH":goto = ENDreturn Command(goto=goto, update={"next": goto})research_agent = create_react_agent(llm, tools=[tavily_tool], state_modifier="You are a researcher. DO NOT do any math."
  )def research_node(state: State) -> Command[Literal["supervisor"]]:result = research_agent.invoke(state)return Command(update={"messages": [HumanMessage(content=result["messages"][-1].content, name="researcher")]},goto="supervisor",)code_agent = create_react_agent(llm, tools=[python_repl_tool])def code_node(state: State) -> Command[Literal["supervisor"]]:result = code_agent.invoke(state)return Command(update={"messages": [HumanMessage(content=result["messages"][-1].content, name="coder")]},goto="supervisor",)builder = StateGraph(State)
  builder.add_edge(START, "supervisor")
  builder.add_node("supervisor", supervisor_node)
  builder.add_node("researcher", research_node)
  builder.add_node("coder", code_node)
  graph = builder.compile()for s in graph.stream({"messages": [("user", "What's the square root of 42?")]}, subgraphs=True
  ):print(s)
  

  

参考智能体元年，四大 AI Agent 框架介绍（LangGraph、CrewAI、AutoGen、LammaIndex）

6、总结：如何通过Chain和Agent，实现更强大的Agent系统

基于LangChain框架的设计，Agents与Chains的结合体现了智能决策与流程化处理的深度融合。

Agents与Chains的结合本质上是​“指挥官与执行部队”​的关系：

• ​Agents：负责高层策略（如任务拆解、工具选择）；
• ​Chains：作为标准化战术单元执行具体操作。
  这种架构既保留了Agent的灵活性，又通过Chain保证了流程的可控性，适用于需兼顾智能决策与流程稳定性的复杂AI应用

以下是其协同运作的核心逻辑与技术实现：

1）动态决策与流程调度的互补机制

Agent主导任务分解 -> 每个子任务对应一个Chain（Chain可复用）

1. ​Agent主导任务分解
   Agent通过意图分析（如用户请求的语义理解）将复杂任务拆解为原子化操作，例如先调用检索工具获取数据，再触发生成链输出结果。此时，每个子任务可能对应一个预定义的Chain（如RetrievalQAChain）。
   示例：用户要求“分析某公司财报并生成投资建议”，Agent可能依次调用：

   • 文档解析链（拆分PDF表格） →

   • 数据清洗链（提取关键指标） →

   • 生成链（结合金融模型输出建议）。

2. ​Chain作为可复用工具单元
   Agent通过工具注册机制（Tool）将Chain封装为标准化接口，例如将LLMChain包装为文本生成工具。当Agent决策需要执行特定步骤时，直接调用这些工具化Chain，实现模块化复用

2）多级流程编排与状态传递

1. ​嵌套链式执行
   Agent可触发多级Chain嵌套，例如在SequentialChain中串联“检索→分析→生成”流程，并通过input_variables自动传递上下文参数（如中间结果）。这种设计避免了手动管理中间状态。 技术实现：

   agent.run({"input": "用户问题","chains": [retrieval_chain, analysis_chain, generation_chain]
   })
   

2. ​记忆系统的双向同步
   Agent的会话历史（ConversationBufferMemory）与Chain的上下文缓存（如ConversationalRetrievalChain的记忆模块）双向同步。例如，在对话场景中，Agent通过记忆库获取历史交互数据，将其注入Chain的提示模板，确保生成内容连贯

3）智能路由与条件分支控制

1. ​动态链选择（RouterChain）​
   Agent根据输入特征（如用户意图分类结果）触发不同的Chain分支。例如：

   • 若检测到“技术文档查询”意图 → 调用RetrievalQAChain；

   • 若检测到“代码生成”意图 → 调用CodeGenerationChain。

2. ​工具链的混合调用
   Agent可同时协调外部API工具（如数据库查询）与Chain。例如在电商场景中，Agent可能：

   • 调用订单查询API →
   • 使用TransformChain格式化数据 →
   • 触发LLMChain生成客户回复

4）异常处理与流程监控

1. ​错误重试与降级策略
   当Chain执行失败（如API超时），Agent的AgentExecutor可自动重试或切换备用链（如从GPT-4降级至本地模型链）。

   agent = initialize_agent(tools=[chain_tool],llm=llm,max_retries=3,fallback_chain=local_chain
   )
   

2. ​回调系统的集成
   通过Callbacks模块监控Chain的执行状态（如耗时、资源消耗），实时反馈至Agent的决策逻辑。例如，若检测到检索链响应延迟过高，Agent可动态切换至缓存策略

6）典型应用场景

1. ​多模态任务处理
   Agent调用视觉处理链（如图像描述生成）与文本生成链协作，实现“图片分析→报告生成”的端到端流程。

2. ​实时数据管道
   在物联网场景中，Agent通过传感器数据采集链获取实时数据，触发预警链生成通知，并记录至数据库链。

三、开发模板和注意事项

开发注意事项：

开发步骤	关键操作	注意事项
​1. 环境配置	安装核心包（langchain-core）、模型接口包（如langchain-openai）	使用虚拟环境隔离依赖；通过环境变量管理API密钥防止泄露
​2. 组件选择	选择LLM（速度/质量权衡）、记忆策略（短期/长期存储）、工具（API/数据库）	优先官方集成工具（如langchain-community）；评估模型Token限制
​3. 链/代理设计	简单任务用LLMChain，多步骤任务用SequentialChain；动态决策场景用代理	使用LCEL优化异步处理；通过langsmith调试执行路径
​4. 测试部署	单元测试验证输出结构；集成测试模拟用户交互 → 使用langserve发布为API	压力测试高并发场景；监控API响应延迟与错误率
​5. 性能调优	启用模型缓存减少延迟；优化分块策略（重叠窗口/语义分割）；索引分片提升检索效率	避免长文本超出模型上下文限制；使用ConversationSummaryMemory压缩历史

典型案例参考：

案例类型	实现方案	技术组合
​文档问答系统	UnstructuredPDFLoader加载文档 → RecursiveCharacterTextSplitter分块 → RetrievalQAChain生成答案	向量存储选用ChromaDB；输出解析器转Markdown
​多语言翻译器	RouterChain识别语言 → 调用对应翻译链 → ConversationBufferMemory保存偏好	使用PromptTemplate定制语言风格；代理调用Google翻译API

1、模型 I/O 开发模板：管理大模型输入输出，包含提示模板、模型调用接口、输出解析器

1）模型加载模板：通过OpenAI兼容接口调用

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate'''
方法1: 通过OpenAI兼容接口调用
'''
# 配置API参数
API_KEY = "sk-82b41f672c5848c18eb67df761952a46"  # 替换为硅基流动密钥，即token
# BASE_URL = "https://api.siliconflow.cn/v1/chat/completions"  # 固定接口地址
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"  # 固定接口地址# 初始化自定义模型
model = ChatOpenAI(api_key=API_KEY,base_url=BASE_URL,model="qwen-turbo"  # 根据需求选择模型ID/code(@ref)
)# 示例：构建问答链
if __name__ == '__main__':prompt = ChatPromptTemplate.from_template("回答：{question}")chain = prompt | modelresponse = chain.invoke({"question": "量子计算的优势是什么？"})print(response.content)

2）硅基流动LLM加载：在LangChain中集成自定义模型

'''
方法3: 在LangChain中集成自定义模型
'''
# 设置API密钥和基础URL环境变量
API_KEY = os.getenv("CUSTOM_API_KEY", "sk-eebarwnbkslesspekwxjbongprpoxdftnhmswtjxmlmyvzgp")
BASE_URL = "https://api.siliconflow.cn/v1/chat/completions"class SiliconFlow(LLM):def __init__(self):super().__init__()@propertydef _llm_type(self) -> str:return "siliconflow"def siliconflow_completions(self, model: str, prompt: str) -> str:payload = {"model": model,"messages": [{"role": "user", "content": prompt}],"stream": False}headers = {"accept": "application/json","content-type": "application/json","authorization": f"Bearer {API_KEY}"}response = requests.post(BASE_URL, json=payload, headers=headers)response.raise_for_status()return response.json()["choices"][0]["message"]["content"]def _call(self, prompt: str, stop: list = None, model: str = "default-model") -> str:response = self.siliconflow_completions(model=model, prompt=prompt)if stop is not None:response = enforce_stop_tokens(response, stop)return responseif __name__ == "__main__":llm = SiliconFlow()response = llm._call(prompt="你是谁？", model="deepseek-ai/DeepSeek-R1-Distill-Qwen-7B")print(response)

2、数据连接 开发模板：实现外部数据加载、向量化存储与检索

1）流程梳理：文档分块、向量化入库、检索与召回

文档分块处理（文档加载 -> 文档分块） -> 文档分块后的向量化存储（加载词嵌入模型 -> 通过绑定词嵌入模型、创建指定集合名称的向量库 ）-> 文档检索和召回（待查询文本 -> 使用指定相似度方法，创建检索器 -> 检索并召回相似度高的文本）

流程说明：

1. 文档分块处理 子流程：

   • 从原始文档加载开始
   • 通过分块算法将文档拆分为语义单元

2. 向量化存储 子流程：

   • 加载预训练的词嵌入模型（如BERT）
   • 将模型绑定到向量数据库
   • 创建带集合名称的向量库（如ChromaDB集合）

3. 文档检索和召回 子流程：

   • 接收查询文本输入
   • 基于余弦相似度/欧氏距离等方法创建检索器
   • 从向量库召回最相似的文档块

各子流程通过箭头连接形成完整pipeline，符合典型的RAG（检索增强生成）系统架构。

2）模块代码模板：分块策略、嵌入模型、向量库、检索器

a）分块策略

from abc import ABC, abstractmethod
from langchain_core.documents import Document'''
文档分块处理方法包括：RecursiveCharacterTextSplitter
'''
class BaseSplitter(ABC):@abstractmethoddef process(self, file_path: str) -> list[Document]:passclass RecursiveTextSplitter(BaseSplitter):def __init__(self, chunk_size=1000, chunk_overlap=200):from langchain_text_splitters import RecursiveCharacterTextSplitterself.splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap,length_function=len,is_separator_regex=False)def process(self, file_path: str) -> list[Document]:from langchain_community.document_loaders import PyPDFLoaderloader = PyPDFLoader(file_path)return self.splitter.split_documents(loader.load())

b）嵌入模型

import os
from langchain_community.embeddings import HuggingFaceEmbeddingsclass EmbeddingModelFactory:@staticmethoddef create_embedding(model_type: str, model_path: str):"""支持多种中文优化模型"""common_params = {"encode_kwargs": {'normalize_embeddings': True},"model_kwargs": {'device': 'cpu'}}if model_type == "m3e-base":return HuggingFaceEmbeddings(model_name=model_path,**common_params)elif model_type == "text2vec":return HuggingFaceEmbeddings(model_name="GanymedeNil/text2vec-large-chinese",**common_params)elif model_type == "bge-small-zh":return HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5",**common_params)else:raise ValueError(f"Unsupported model type: {model_type}")

c）向量库

from abc import ABC, abstractmethodfrom langchain_core.documents import Document
from langchain_core.embeddings import Embeddings'''
向量数据库包括：Chroma
'''
class VectorStore(ABC):def __init__(self, persist_dir: str, collection: str):self.persist_dir = persist_dirself.collection = collection@abstractmethoddef store(self, chunks: list[Document], embedding_model: Embeddings):passclass ChromaStore(VectorStore):def store(self, chunks, embedding_model):from langchain_chroma import Chromareturn Chroma.from_documents(documents=chunks,embedding=embedding_model,persist_directory=self.persist_dir,collection_name=self.collection)

d）检索器

from abc import ABC, abstractmethodfrom langchain_core.documents import Document
from langchain_core.vectorstores import VectorStoreRetrieverclass BaseRetriever(ABC):@abstractmethoddef search(self, query: str, **kwargs) -> list[Document]:passclass RetrieverFactory:@staticmethoddef create_retriever(retriever_type: str, vector_store, **kwargs):"""支持三种检索器类型：mmr, similarity, threshold"""if retriever_type == "mmr":return MMRRetriever(vector_store, **kwargs)elif retriever_type == "similarity":return CosineSimilarityRetriever(vector_store, **kwargs)elif retriever_type == "threshold":return ThresholdRetriever(vector_store, **kwargs)raise ValueError(f"Unsupported retriever type: {retriever_type}")class MMRRetriever(BaseRetriever):def __init__(self, vector_store, k: int = 5, lambda_mult: float = 0.5):self.retriever = vector_store.as_retriever(search_type="mmr",search_kwargs={"k": k, "lambda_mult": lambda_mult})def search(self, query: str, **kwargs) -> list[Document]:return self.retriever.invoke(query)class CosineSimilarityRetriever(BaseRetriever):def __init__(self, vector_store, k: int = 5):self.retriever = vector_store.as_retriever(search_type="similarity",search_kwargs={"k": k})def search(self, query: str, **kwargs) -> list[Document]:return self.retriever.invoke(query)class ThresholdRetriever(BaseRetriever):def __init__(self, vector_store, threshold: float = 0.7, k: int = 5):self.retriever = vector_store.as_retriever(search_type="similarity",search_kwargs={"score_threshold": threshold, "k": k})def search(self, query: str, **kwargs) -> list[Document]:return self.retriever.invoke(query)

e）主流程

import os
from pathlib import Path
from utils.splitter_model import RecursiveTextSplitter
from utils.vector_store import ChromaStore
from utils.retriever_model import MMRRetriever, RetrieverFactory
from utils.embedding_model import EmbeddingModelFactory
from langchain_community.embeddings import HuggingFaceEmbeddings# 配置参数
CONFIG = {"pdf_path": "../../../resources/中华人民共和国公安部令道路交通安全违法行为记分管理办法.pdf","vector_dir": "D:/programSoftware/langChain/vector_storage/chroma","vector_model_path": "D:/programSoftware/langChain/huggingface/hub/models--moka-ai--m3e-base/snapshots/764b537a0e50e5c7d64db883f2d2e051cbe3c64c","collection_name": "traffic_laws"
}def main():# 1. 文档分块print("Step1: 开始分块...")splitter = RecursiveTextSplitter(chunk_size=1000, chunk_overlap=200)chunks = splitter.process(CONFIG["pdf_path"])# 2. 向量化存储print("Step2: 加载本地嵌入模型...")os.environ['TRANSFORMERS_OFFLINE'] = '1'  # 强制离线模式os.environ['HF_DATASETS_OFFLINE'] = '1'  # 数据集也离线（如有）# 创建词嵌入模型实例embeddings = EmbeddingModelFactory.create_embedding(model_type="m3e-base",  # 可配置为其他类型model_path=CONFIG["vector_model_path"])print("Step3: 加载本地向量库...")vector_store = ChromaStore(persist_dir=CONFIG["vector_dir"],collection=CONFIG["collection_name"]).store(chunks, embeddings)# 3. 检索测试print("Step4: 加载检索器...")# MMR检索示例retriever = RetrieverFactory.create_retriever("mmr",vector_store,k=5,lambda_mult=0.6  # 相关性/多样性平衡系数)print("Step5: 检索结果...")results = retriever.search("公安机关交通管理部门")print(f"检索到 {len(results)} 条结果")# 输出结果print(f"存储路径：{list(Path(CONFIG['vector_dir']).glob('*'))}")print("\n检索结果：")for doc in results:print(doc.page_content[:200] + "...")if __name__ == "__main__":main()

3）检索结果可视化

方法	关键参数	推荐值	作用说明
PCA	n_components	2/3	降维目标维度
t-SNE	perplexity	5-50	控制邻域大小，值越大考虑更多全局结构
UMAP	n_neighbors	15-50	平衡局部与全局结构
HDBSCAN	min_cluster_size	5-20	控制最小簇大小

4）检索效果评估

3、链（Chains）开发模板：串联多个处理步骤形成端到端任务流

1）流程梳理：定义提示词模板、根据场景创建多类型链并执行

关于"链"的开发，关键流程如下：LLM模型加载 -> 定义提示词模板（定义输入变量） -> 创建链（根据场景选择链的类型） -> 执行链，获取结果

关键说明：

1. ​模型加载：通过 ChatOpenAI 加载硅基流动的 DeepSeek-R1-Distill-Qwen-7B 模型，支持自定义API参数

2. ​提示模板：

   • ​领域模板：physics_prompt_tpl 和 math_prompt_tpl 分别定义物理和数学问题的回答规范。

   • ​路由模板：整合所有子链描述，生成动态路由决策提示（如 physics: 用于解答物理相关问题）

3. ​链创建：

   • ​目标链：通过 LLMChain 为每个模板创建专用处理链（如 physics_chain）

   • ​路由链：LLMRouterChain 解析输入并选择目标链，RouterOutputParser 处理路由结果

   • ​默认链：ConversationChain 处理未匹配输入，集成 ConversationBufferMemory 维护对话历史

4. ​链执行：通过 MultiPromptChain 整合路由逻辑和子链，调用 run() 方法执行完整流程。

关于LLMChain、TransformChain、SequentialChain、ConversationChain、routerChain、MultiPromptChain的核心Chain类型对比表如下：

Chain类型	核心功能	是否调用LLM	典型应用场景
​LLMChain	基础链，直接调用LLM处理单一任务	✅	简单问答、文本生成
​TransformChain	数据预处理链，执行纯数据转换	❌	输入格式标准化、文本分块（比如文本分块预处理），输出为字典
​SequentialChain	顺序执行链，将多个链组合成线性流程	可选	多步骤任务（比如文档摘要：分块→摘要生成→结果整合）
​ConversationChain	对话链，集成记忆管理模块	✅	多轮对话系统
​RouterChain	路由决策链，动态选择子链执行路径	可选	领域分类路由
​MultiPromptChain	多提示链，RouterChain的子类，专用于多模板场景	✅	多专家系统

选型建议：

1. ​简单任务：使用LLMChain
2. ​数据清洗 ： 使用TransformChain（无需调用LLM）
3. ​多步骤流程：使用 SequentialChain（需状态传递时）或使用LangGraph（复杂DAG时）
4. ​上下文对话：使用 ConversationChain（集成记忆管理）
5. ​领域分流 ： 使用MultiPromptChain（预置模板）或自定义RouterChain（需灵活路由逻辑）

2）提示词模板：自定义模板变量 & 流水线复杂模板

最佳实践：

• 模块化设计 ：将复杂模板拆分为子模板，通过 PipelinePrompt 组合。
• 验证变量 ：确保所有 input_variables 都在调用时提供。
• 使用 partial 方法 ：固定通用参数（如语言、格式）。
• 测试边缘情况 ：验证模板在缺失变量或异常输入

a）单模板（PromptTemplate）：变量 & 部分固定变量

Note - 使用PromptTemplate定义模板

定义多变量模板：

print("1. 定义多变量模板")
prompt_template = PromptTemplate(input_variables=["topic", "audience", "tone"],template="""请根据以下要求生成内容：- 主题：{topic}- 目标读者：{audience}- 语气风格：{tone}"""
)

定义部分固定变量：

print("2. 定义部分固定变量")
prompt_template = PromptTemplate(input_variables=["question"],partial_variables={"language": "中文", "max_length": "500字"},template="语言：{language}\n长度限制：{max_length}\n问题：{question}"
)

b）流水线模板（PipelinePromptTemplate）

Note - 使用PipelinePromptTemplate组合多个PromptTemplate模板

# 定义基础模板
print(f"2. 定义基础模板")
base_prompt = PromptTemplate(input_variables=["topic"],template="请解释以下概念：{topic}"
)# 定义修饰模板
print(f"3. 定义修饰模板")
decorator_prompt = PromptTemplate(input_variables=["style"],template="要求：用{style}风格回答"
)# 定义最终模板（合并子模板的输出）
print(f"4. 定义最终模板")
final_template = PromptTemplate(input_variables=["base","decorator"],template="""{base}\n{decorator}"""
)# 组合成流水线
print(f"5. 组合成流水线模板")
pipeline_prompt = PipelinePromptTemplate(pipeline_prompts=[("base", base_prompt),("decorator", decorator_prompt)],final_prompt=final_template
)

3）多类型链：​LLMChain、TransformChain、SequentialChain、ConversationChain、RouterChain、MultiPromptChain

a）简单链（LLMChain） & 转换链（TransformChain）

# 定义多变量模板
prompt_template = PromptTemplate(input_variables=["topic", "audience", "tone"],template="""请根据以下要求生成内容：- 主题：{topic}- 目标读者：{audience}- 语气风格：{tone}"""
)# 3. 构建简单链
simple_chain = LLMChain(llm=llm, prompt=prompt_template)

b）顺序执行链（SequentialChain）

def clean_data(inputs):raw_data = inputs["raw_data"]cleaned_data = raw_data.replace("错误值", "默认值")  # 自定义清洗逻辑return {"cleaned_data": cleaned_data}print("Step1: 定义数据清洗转换链: 输入:raw_data，输出:cleaned_data")
transform_chain = TransformChain(input_variables=["raw_data"],output_variables=["cleaned_data"],transform=clean_data
)# 2. 定义分析链（LLM 生成报告）
print("Step3: 定义分析链（LLM 生成报告）: 输入:cleaned_data, 输出: 分析报告")
prompt = PromptTemplate(input_variables=["cleaned_data"],template="根据以下数据生成分析报告：{cleaned_data}"
)
analysis_chain = LLMChain(llm=llm, prompt=prompt, output_key="report")# 3. 组合复杂链
print("Step4: 用SequentialChain定义组合复杂链: 输入:raw_data，输出:report")
complex_chain = SequentialChain(chains=[transform_chain, analysis_chain],input_variables=["raw_data"],output_variables=["report"]
)

c）多轮对话链（ConversationChain）

memory = ConversationBufferMemory()
# 自定义提示模板（显式使用历史记录）
template = """
你是一个对话助手，需结合历史记录回答问题：
历史对话：
{history}
用户当前问题：
{input}
"""
prompt = PromptTemplate(input_variables=["history", "input"],template=template
)conversation = ConversationChain(llm=llm,memory=memory,prompt=prompt,verbose=True  # 查看提示内容是否包含历史记录
)

d）路由链（RouterChain）

physics_prompt_tpl = PromptTemplate(template = """假设你是一个物理老师，你需要根据用户的问题，用20字回答物理相关的问题, 问题如下:{input}。""",input_variables = ["input"]
)math_prompt_tpl = PromptTemplate(template = """假设你是一个数学老师，你需要根据用户的问题，用20字回答数学相关的问题, 问题如下:{input}。""",input_variables = ["input"]
)'''
定义模板信息字典
创建包含各子链特征的列表，每个元素包含：- name：链标识符（如physics/math）- description：功能描述（用于生成路由提示）- prompt_template：专用提示模板
'''
prompt_infos = [{"name": "physics","description": "用于解答物理相关问题","prompt_template": physics_prompt_tpl},{"name": "math","description": "用于解答数学相关问题","prompt_template": math_prompt_tpl}
]'''
生成目标链集合
遍历模板信息，创建LLMChain并存入字典:
'''
print("Step3: 通过LLMChain生成目标链集合...")
destination_chains = {}
for p_info in prompt_infos:chain = LLMChain(llm=llm, prompt=p_info['prompt_template'])destination_chains[p_info['name']] = chain'''
构建路由提示模板: 使用MULTI_PROMPT_ROUTER_TEMPLATE生成路由决策依据：
该模板将子链描述拼接为结构化文本，供LLM判断输入归属
'''
destinations_str = "\n".join([f"{p['name']}: {p['description']}" for p in prompt_infos])
print(f"Step4: 构建路由提示模板:{destinations_str}\n")
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations=destinations_str)'''
初始化路由链与默认链- 路由链：解析LLM的路由决策- 默认链：处理未匹配的输入
'''
print("Step6: 通过LLMRouterChain 初始化 路由链...")
router_prompt = PromptTemplate(template=router_template,input_variables=["input"],output_parser=RouterOutputParser()
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

路由链执行流程：

定义目标链（physics/math/default链，需绑定LLM） -> 根据路由模板（包含目标模板），定义路由链（需绑定LLM） -> 定义默认链 -> 组合路由链和默认链为完整链 -> 执行完整链

输入类型	预期路由目标	测试用例	输出特征
物理问题	physics链	“牛顿第一定律是什么？”	包含物理学专业术语
数学问题	math链	“鸡兔同笼问题解法”	展示数学公式推导
未定义领域问题	default链	“如何烹饪牛排？”	通用对话式回答

e）组合链（MultiPromptChain）

# 省略router_chain的构建...
router_chain = LLMRouterChain.from_llm(llm, router_prompt)# 初始化内存和对话链
memory = ConversationBufferMemory()
# 会报错: Value error, Got unexpected prompt input variables. The prompt expects ['input'], but got ['history'] as inputs from memory, and input as the normal input key. [type=value_error, input_value={'llm': ChatOpenAI(client...    '), 'verbose': True}, input_type=dict]
# common_prompt_tpl = PromptTemplate(
#     template = """
#         回答通识问题, 问题如下:{input}。
#     """,
#     input_variables = ["input"]
# )
template = """
你是一个对话助手，需结合历史记录回答问题：
历史对话：
{history}
用户当前问题，用20字回答：
{input}
"""
common_prompt_tpl = PromptTemplate(input_variables=["history", "input"],template=template
)
default_chain = ConversationChain(llm=llm,memory=memory,prompt=common_prompt_tpl,output_key="text",verbose=True  # 查看提示内容是否包含历史记录
)'''
组合完整处理链
'''
print("Step6: 通过MultiPromptChain组合完整处理链...")
chain = MultiPromptChain(router_chain=router_chain,destination_chains=destination_chains,default_chain=default_chain,verbose=True
)

4）主流程

a）提示词模板类

from langchain.prompts import PromptTemplate, PipelinePromptTemplate'''
提示词模板包括：自定义模板变量 & 流水线复杂模板
'''
class PromptTemplateFactory:@staticmethod# 如果参数不存在则为空def create_template(template_type: str, **kwargs):output_parser = kwargs["output_parser"] if "output_parser" in kwargs.keys() else Noneinput_variables = kwargs["input_variables"] if "input_variables" in kwargs.keys() else Nonepartial_variables = kwargs["partial_variables"] if "partial_variables" in kwargs.keys() else Nonetemplate = kwargs["template"] if kwargs["template"] else None"""支持创建多类型提示模板"""if template_type == "multi_variable":return PromptTemplate(input_variables=input_variables,template=template,output_parser=output_parser)elif template_type == "partial_variable":return PromptTemplate(input_variables=input_variables,partial_variables=partial_variables,template=template,output_parser=output_parser)elif template_type == "pipeline":return PipelinePromptTemplate(pipeline_prompts=kwargs["pipeline_prompts"],final_prompt=kwargs["final_prompt"])raise ValueError(f"Unsupported template type: {template_type}")

b）链式处理器类

from langchain.chains import LLMChain, SequentialChain, MultiPromptChain, ConversationChain
from langchain.chains.router import RouterChain, LLMRouterChain
from langchain.chains.transform import TransformChain'''
链模型包括：LLMChain、TransformChain、SequentialChain、
ConversationChain、RouterChain、MultiPromptChain
'''
class ChainFactory:@staticmethoddef create_chain(chain_type: str, **kwargs):llm = kwargs["llm"] if "llm" in kwargs.keys() else Noneprompt = kwargs["prompt"] if "prompt" in kwargs.keys() else Noneinput_variables = kwargs["input_variables"] if "input_variables" in kwargs.keys() else Noneoutput_variables = kwargs["output_variables"] if "output_variables" in kwargs.keys() else Noneoutput_key = kwargs["output_key"] if "output_key" in kwargs.keys() else "text"transform = kwargs["transform"] if "transform" in kwargs.keys() else None"""链式处理器工厂"""# 简单链if chain_type == "simple":print("创建简单链")return LLMChain(llm=llm,prompt=prompt,output_key=output_key)# 转换链elif chain_type == "transform":print("创建转换链")return TransformChain(input_variables=input_variables,output_variables=output_variables,transform=transform)# 多步骤执行链elif chain_type == "sequential":print("创建多步骤执行链")return SequentialChain(chains=kwargs["chains"],input_variables=input_variables,output_variables=output_variables)# 多轮对话链elif chain_type == "conversation":print("创建多轮对话链")return ConversationChain(llm=llm,memory=kwargs["memory"],prompt=prompt,output_key = output_key,verbose = kwargs["verbose"],  # 查看提示内容是否包含历史记录)# 路由链elif chain_type == "router":print("创建路由链")return RouterChain(llm=llm,router_chain=kwargs["router_chain"],destination_chains=kwargs["destination_chains"],default_chain=kwargs["default_chain"])elif chain_type == "llm_router":print("创建路由链")return LLMRouterChain.from_llm(llm=kwargs["llm"], prompt=kwargs["prompt"])# 组合链elif chain_type == "multi_prompt":print("创建组合链")return MultiPromptChain(router_chain=kwargs["router_chain"],destination_chains=kwargs["destination_chains"],default_chain=kwargs["default_chain"])raise ValueError(f"Unsupported chain type: {chain_type}")

b）主函数

场景描述：

• 用sequentialChain定义分析链，生成数据分析报告

• 用routerChain定义路由链，分配给不同专家来回答问题

• 用MultiPromptChain定义组合链，若找不到专家则使用通用专家来回答问题

from langchain.chains.router.llm_router import RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
from langchain.memory import ConversationBufferMemoryfrom utils.llm_loader import LLMLoader
from utils.prompt_template_model import PromptTemplateFactory
from utils.chain_model import ChainFactoryCONFIG = {"api_key": "sk-eebarwnbkslesspekwxjbongprpoxdftnhmswtjxmlmyvzgp","base_url": "https://api.siliconflow.cn/v1","model_id": "deepseek-ai/DeepSeek-R1-Distill-Qwen-7B"
}def clean_data(inputs):raw_data = inputs["raw_data"]cleaned_data = raw_data.replace("错误值", "默认值")  # 自定义清洗逻辑return {"cleaned_data": cleaned_data}# 创建多步骤执行链
def create_sequence_chain(llm):# 1. 定义转换链transform_chain = ChainFactory.create_chain(chain_type="transform",input_variables=["raw_data"],output_variables=["cleaned_data"],transform=clean_data)# 2. 定义分析链（LLM 生成报告）prompt = PromptTemplateFactory.create_template(template_type="multi_variable",input_variables=["cleaned_data"],template="根据以下数据生成分析报告：{cleaned_data}")analysis_chain = ChainFactory.create_chain(chain_type="simple",llm=llm,prompt=prompt,output_key="report")# 3. 组合复杂链print("Step4: 用SequentialChain定义组合复杂链: 输入:raw_data，输出:report")complex_chain = ChainFactory.create_chain(chain_type="sequential",chains=[transform_chain, analysis_chain],input_variables=["raw_data"],output_variables=["report"])return complex_chain# 创建会话链
def create_conversation_chain(llm):memory = ConversationBufferMemory()# 会报错: Value error, Got unexpected prompt input variables. The prompt expects ['input'], but got ['history'] as inputs from memory, and input as the normal input key. [type=value_error, input_value={'llm': ChatOpenAI(client...    '), 'verbose': True}, input_type=dict]# common_prompt_tpl = PromptTemplate(#     template = """#         回答通识问题, 问题如下:{input}。#     """,#     input_variables = ["input"]# )template = """你是一个对话助手，需结合历史记录回答问题：历史对话：{history}用户当前问题，用20字回答：{input}"""common_prompt_tpl = PromptTemplateFactory.create_template(template_type="multi_variable",input_variables=["history", "input"],template=template)default_chain = ChainFactory.create_chain(chain_type="conversation",llm=llm,memory=memory,prompt=common_prompt_tpl,output_key="text",verbose=True  # 查看提示内容是否包含历史记录)return default_chain# 创建路由链
def create_route_chain(llm):physics_prompt_tpl = PromptTemplateFactory.create_template(template_type="multi_variable",template="""假设你是一个物理老师，你需要根据用户的问题，用20字回答物理相关的问题, 问题如下:{input}。""",input_variables=["input"])math_prompt_tpl = PromptTemplateFactory.create_template(template_type="multi_variable",template="""假设你是一个数学老师，你需要根据用户的问题，用20字回答数学相关的问题, 问题如下:{input}。""",input_variables=["input"])'''定义模板信息字典创建包含各子链特征的列表，每个元素包含：- name：链标识符（如physics/math）- description：功能描述（用于生成路由提示）- prompt_template：专用提示模板'''prompt_infos = [{"name": "physics","description": "用于解答物理相关问题","prompt_template": physics_prompt_tpl},{"name": "math","description": "用于解答数学相关问题","prompt_template": math_prompt_tpl}]'''生成目标链集合遍历模板信息，创建LLMChain并存入字典:'''print("Step3: 通过LLMChain生成目标链集合...")destination_chains = {}for p_info in prompt_infos:chain = ChainFactory.create_chain(chain_type = "simple",llm=llm,prompt=p_info['prompt_template'])destination_chains[p_info['name']] = chain'''构建路由提示模板: 使用MULTI_PROMPT_ROUTER_TEMPLATE生成路由决策依据：该模板将子链描述拼接为结构化文本，供LLM判断输入归属'''destinations_str = "\n".join([f"{p['name']}: {p['description']}" for p in prompt_infos])print(f"Step4: 构建路由提示模板:{destinations_str}\n")router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations=destinations_str)router_prompt = PromptTemplateFactory.create_template(template_type = "multi_variable",template=router_template,input_variables=["input"],output_parser=RouterOutputParser())router_chain = ChainFactory.create_chain(chain_type="llm_router",llm=llm,prompt=router_prompt)return router_chain,destination_chainsdef main():# 初始化模型加载器print("Step1: 初始化模型加载器...")loader = LLMLoader(CONFIG["api_key"], CONFIG["base_url"])llm = loader.load_model(CONFIG["model_id"])# 初始化默认执行的处理链print("Step2: 初始化默认执行的处理链...")sequence_chain = create_sequence_chain(llm)result = sequence_chain({"raw_data": "2023年销售额：错误值，2024年预测：增长20%"})# 流式输出print(result["report"])print("Step3: 组合完整处理链...")# 初始化路由链router_chain,destination_chains = create_route_chain(llm)# 初始化会话链default_chain = create_conversation_chain(llm)multi_prompt_chain = ChainFactory.create_chain(chain_type = "multi_prompt",router_chain=router_chain,destination_chains=destination_chains,default_chain=default_chain,verbose=True)print("Step7: 执行链...")# result = chain.run("math:2+2等于几？")result = multi_prompt_chain.run(input="数学:2+2等于几？")print(result)# result = chain.run("physics:浮力定律是什么？")result = multi_prompt_chain.run(input="物理:浮力定律是什么？")print(result)# 报错:result = multi_prompt_chain.run(input="biology:生物细胞有哪些器官？")print(result)if __name__ == "__main__":main()

执行结果：

Step1: 初始化模型加载器...
Step2: 初始化默认执行的处理链...
创建转换链
创建简单链
Step4: 用SequentialChain定义组合复杂链: 输入:raw_data，输出:report
创建多步骤执行链
### 销售预测分析报告#### 1. 背景说明
2023年的销售情况是基于当前市场、客户和公司策略的综合分析得出，默认值的2023年销售额被视为基准。2024年预测增长20%，这一增长目标设定为基于核心业务的稳健发展、市场扩张以及公司内部策略的优化。以下将详细分析预测的依据和可能的影响因素。#### 2. 销售预测
- **2023年销售额**：按照基准值，假设为X Million单位。
...
...
2+2等于4。
浮力定律：物体排开流体的重量等于浮力。> Entering new ConversationChain chain...
Prompt after formatting:你是一个对话助手，需结合历史记录回答问题：历史对话：用户当前问题，用20字回答：biology:生物细胞有哪些器官？> Finished chain.
生物细胞主要有细胞膜、细胞质、线粒体、叶绿体、核糖体、核、内质网、高尔基体、液泡等结构。

4、记忆（Memory）开发模板：维护对话历史或任务上下文

1）流程梳理：初始化内存/外存 -> 读取或检索数据

短期记忆：初始化LLM模型 -> 初始化内存（设置滑动窗口） -> 基于内存创建多轮对话链 -> 多轮对话，流程图如下：

长期记忆：初始化Embedding模型（如果需要初始化向量库） -> 初始化外存（向量库 / Redis / MySQL / KG）-> 存储数据到外存 -> 从外存中读取数据（检索召回/调用查询）

2）代码模板：短期记忆，长期记忆，混合策略

a）短期记忆

'''
短期记忆实现（基于内存）
ConversationBufferMemory
- return_messages=True：以消息对象格式存储对话
'''
# 初始化模型和内存
memory = ConversationBufferMemory(return_messages=True)
chain = ConversationChain(llm=llm, memory=memory)# 测试多轮对话
response1 = chain.invoke({"input": "我叫李雷，职业是医生"})["response"]  # 存储姓名和职业
print(f"第一轮输出:{response1}")  # 输出：您叫李雷，职业是医生
response2 = chain.invoke({"input": "中国最高的山是哪座？"})["response"]  # 无关问题
print(f"第二轮输出:{response2}")  # 输出：您叫李雷，职业是医生
response3 = chain.invoke({"input": "我的名字是什么？"})["response"]  # 正确回忆姓名
print(f"第三轮输出:{response3}")  # 输出：您叫李雷，职业是医生

控制滑动窗口来保留对话轮次

'''
进阶版：基于内存的带滑动窗口的记忆
ConversationBufferWindowMemory
- k=2：控制滑动窗口保留的对话轮次
'''
# 只保留最近2轮对话
window_memory = ConversationBufferWindowMemory(k=2, return_messages=True)
chain = ConversationChain(llm=llm, memory=window_memory)response1 = chain.invoke({"input": "我的名字是韩梅梅"})["response"]
print(f"第一轮输出:{response1}")  # 输出：您叫李雷，职业是医生
response2 = chain.invoke({"input": "我今年25岁"})["response"]
print(f"第一轮输出:{response2}")  # 输出：您叫李雷，职业是医生
response3 = chain.invoke({"input": "我的名字是什么？"})["response"]  # 输出：您的名字是韩梅梅
print(f"第一轮输出:{response3}")  # 输出：您叫李雷，职业是医生
response4 = chain.invoke({"input": "我几岁了？"})["response"]  # 输出：您今年25岁
print(f"第一轮输出:{response4}")  # 输出：您叫李雷，职业是医生
response5 = chain.invoke({"input": "我的职业是什么？"})["response"]  # 输出：未提及职业信息（超出窗口限制）
print(f"第一轮输出:{response5}")  # 输出：您叫李雷，职业是医生
response6 = chain.invoke({"input": "我的名字是什么？"})["response"]  # 输出：未提及职业信息（超出窗口限制）
print(f"第一轮输出:{response6}")  # 输出：您叫李雷，职业是医生

b）长期记忆

'''
长期记忆实现（基于外部存储）
向量数据库记忆（Chroma）
'''
# 加载本地中文Embedding模型
print("加载 m3e-base本地模型...")
os.environ['TRANSFORMERS_OFFLINE'] = '1'  # 强制离线模式
os.environ['HF_DATASETS_OFFLINE'] = '1'  # 数据集也离线（如有）
embeddings = HuggingFaceEmbeddings(model_name="D:/programSoftware/langChain/huggingface/hub/models--moka-ai--m3e-base/snapshots/764b537a0e50e5c7d64db883f2d2e051cbe3c64c",  # 建议使用绝对路径encode_kwargs={'normalize_embeddings': True}
)# 初始化向量数据库
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
# search_kwargs={"k":3}：检索最相关的3条记忆
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
memory = VectorStoreRetrieverMemory(retriever=retriever)# 存储长期记忆
memory.save_context({"input": "我的身份证号是110101199003077832"},{"output": "已记录身份证信息"})
memory.save_context({"input": "我的银行账户是6225880134567890"},{"output": "已记录银行账户"})# 检索记忆
print(memory.load_memory_variables({"input": "我的身份证号是多少？"}))
# 输出：包含身份证号的相关上下文

c）混合策略

3）主流程

5、代理（Agents）开发模板：ReAct 和 Function Calling & 动态调用外部工具（API / 数据库）

1）流程梳理：

参考 https://python.langchain.com/docs/how_to/custom_tools/

a）ReAct Agent

创建llm -> 定义只有单个参数的Tool -> 基于ReAct（Zero/Few - Shot）创建Agent -> 执行Agent,处理用户问题

b）Structured Chat

创建llm -> 定义包含多个参数的Tool -> 基于结构化工具(Structured Chat)来创建Agent -> 执行Agent,处理用户问题

c）bind tools

创建llm -> 将Tool绑定到llm上创建Agent -> 执行Agent,处理用户问题

d）PlanAndExecute

创建llm -> 定义只有单个参数的Tool -> 创建计划者和执行者 -> 基于计划者和执行者创建Agent -> 执行Agent,处理用户问题

2）代码模板：

a）ReAct Agent框架：支持单个参数输入 & ZeroShot

使用来@tool()或者StructuredTool.from_function来声明Tool；

使用initialize_agent或create_react_agent，来创建基于ReAct框架的Agent

"""这种定义方式，只能接受一个参数，且不能封装成类，要不然会报少参数"""
print("Step1: 这种定义方式，只能接受一个参数，且不能封装成类，要不然会报少参数...")
class SearchInput(BaseModel):query: str = Field(description="search query")@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:"""Look up things online."""return "我是一个搜索的工具"# print(search.name)
# print(search.description)
# print(search.args)
# print(search.return_direct)"""这种定义方式，可以接受多个参数"""
class CalculatorInput(BaseModel):a: str = Field(description="第一个数字")b: str = Field(description="第二个数字")class SortList(BaseModel):num: str = Field(description="待排序列表")def dort_fun(num):"""Multiply two numbers."""return sorted(eval(num))print("Step2: 这种定义方式，可以接受多个参数...")
sorter = StructuredTool.from_function(func=dort_fun,  # 工具具体逻辑name="sort_num",  # 工具名description="排序列表中的数字",  # 工具信息args_schema=SortList,  # 工具接受参数信息return_direct=True,  # 直接作为工具的输出返回给调用者handle_tool_error=True,  # 报错了继续执行，不会吧那些报错行抛出，也可以自定义函数处理，handle_tool_error=函数名
)"""
zero-shot-react-description
此代理使用ReAct框架，仅基于工具的描述来确定要使用的工具。
可以提供任意数量的工具。
此代理需要为每个工具提供描述
工具只能接受一个参数，不支持多个参数
"""
print("Step3: 使用initialize_agent初始化ReAct框架的Agent: zero-shot-react-description, 此代理使用ReAct框架，仅基于工具的描述来确定要使用的工具")
# llm-math需要安装numexpr包，但linux编译无效，这里就不细究这个问题了
# tools = load_tools(["llm-math"], llm=llm)
# tools = [search, sorter] + tools
tools = [search, sorter]
# initialize agent会报错：AttributeError: type object 'SearchInput' has no attribute 'model_json_schema'
agent = initialize_agent(tools, llm,agent = AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True)
# agent调用工具不一定会生效（没有Observation到）
agent.run("计算 2 ** 3 + 4，`[10,4,7]`排一下序，并查找百度百科中关于'量子计算'的词条")

如果 “tool调用决策模块” 未成功触发调用tool逻辑，则未触发T - A - O逻辑，直接输出文本

> Entering new AgentExecutor chain...langchian refers to the langchian language, and the string "阿萨德防守打法" has 13 characters.Final Answer: "langchian" is the langchian language, and the string "阿萨德防守打法" consists of 13 characters.> Finished chain.Process finished with exit code 0

如果 “tool调用决策模块” 成功触发调用tool逻辑，则ReAct框架的Agent参考输出如下：

1. 初始化自定义模型
2. 定义工具集
3. 使用标准ReAct模板
/root/anaconda3/envs/ga_assistant_env/lib/python3.10/site-packages/langsmith/client.py:277: LangSmithMissingAPIKeyWarning: API key must be provided when using hosted LangSmith API
warnings.warn(
4. 注入工具元数据
5. 创建代理
6. 配置执行器
/tmp/pycharm_project_743/llm_langchain/template/5_Agent代理/2_ReAct代理和工具集成/2_simple_agent1.py:49: LangChainDeprecationWarning: Please see the migration guide at: https://python.langchain.com/docs/versions/migrating_memory/
memory=ConversationBufferMemory(memory_key="chat_history"),  # 明确指定记忆键
6. 正确调用方式
/root/anaconda3/envs/ga_assistant_env/lib/python3.10/site-packages/langchain/memory/chat_memory.py:55: UserWarning: 'ConversationBufferMemory' got multiple output keys: dict_keys(['output', 'intermediate_steps']). The default 'output' key is being used. If this is not desired, please manually set 'output_key'.
warnings.warn(
result = {'input': "我需要查询百度百科的'量子计算'词条",'chat_history': '','output': 'Agent stopped due to iteration limit or time limit.','intermediate_steps': {
'input': "我需要查询百度百科的'量子计算'词条",
'chat_history': '',
'output': 'Agent stopped due to iteration limit or time limit.',
'intermediate_steps': [(AgentAction(tool = '_Exception', tool_input = "Invalid Format: Missing 'Action:' after 'Thought:'", log = '\n\n我决定使用baike_search_tool工具来查找“量子计算”的词条内容。这将帮助我获取关于量子计算的详细信息。\n\nAction Input: quantum computing\n'), "Invalid Format: Missing 'Action:' after 'Thought:'"), (AgentAction(tool = 'baike_search_tool', tool_input = 'quantum computing\n', log = '\n\nThought: 我需要使用baike_search_tool工具来查找关于“量子计算”的词条内容。操作步骤是输入关键词“量子计算”。\n\nAction: baike_search_tool\n\nAction Input: quantum computing\n'), "【quantum computing\n】百科摘要：\n这是接口关于'quantum computing\n'的模拟数据\n"), ...(AgentAction(tool = 'baike_search_tool', tool_input = 'quantum computing', log = "\n\nQuestion: 我需要查询百度百科的'量子计算'词条\nThought: 我决定使用baike_search_tool工具来查找关于“量子计算”的词条内容，并确保正确的格式和思考流程。\nAction: baike_search_tool  \nAction Input: quantum computing  "), "【quantum computing】百科摘要：\n这是接口关于'quantum computing'的模拟数据\n")
]
}

b）Structured Chat: 调用的工具支持多个参数输入

参考 langchain 的 agent + tool 使用

参考 LangChain中文网文档 - Agents类型 - 会话（Conversational）

使用来@tool()或者StructuredTool.from_function来声明Tool；

使用create_structured_chat_agent来创建

class CalculatorInput(BaseModel):a: str = Field(description="第一个字符串")b: str = Field(description="第二个字符串")def multiply(a: str, b: str) -> int:"""Multiply two numbers."""return len(a) * len(b)calculator = StructuredTool.from_function(func=multiply,  # 工具具体逻辑name="Calculator",  # 工具名description="计算字符长度的乘积",  # 工具信息args_schema=CalculatorInput,  # 工具接受参数信息return_direct=True,  # 直接作为工具的输出返回给调用者handle_tool_error=True,  # 报错了继续执行，不会吧那些报错行抛出，也可以自定义函数处理，handle_tool_error=函数名
)class SearchInput(BaseModel):query: str = Field(description="should be a search query")@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:"""Look up things online."""return "你好啊"tools = [search, calculator]
prompt = hub.pull("hwchase17/structured-chat-agent")
print("test1: 使用create_structured_chat_agent创建agent")
agent = create_structured_chat_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True
)
final_result = agent_executor.invoke({"input": "`asd`的字符串长度乘以`as`的字符串长度是多少？langchain是什么？"})
print(f"result:{final_result}")

输出如下，这里没有触发工具调用，说明在Function Calling中间过程，在LLM的输出结果时，并没有转换成"工具调用决策模块"能识别的Schema，导致工具调用决策模块失效：

> Entering new AgentExecutor chain...{"action": "Final Answer","action_input": "将'asd'的字符串长度乘以'as'的字符串长度，得到结果，然后解释LangChain是什么。"
}Finished chain.
result:{'input': '`asd`的字符串长度乘以`as`的字符串长度是多少？langchain是什么？', 'output': "将'asd'的字符串长度乘以'as'的字符串长度，得到结果，然后解释LangChain是什么。"}

c）使用bind_tools，将工具绑定到llm上

参考 https://zhuanlan.zhihu.com/p/718332876

使用来@tool()或者StructuredTool.from_function来声明Tool；

使用llm.bind_tools，将工具绑定到llm上来创建Agent

class CalculatorInput(BaseModel):a: str = Field(description="第一个字符串")b: str = Field(description="第二个字符串")def multiply(a: str, b: str) -> int:"""Multiply two numbers."""return len(a) * len(b)calculator = StructuredTool.from_function(func=multiply,  # 工具具体逻辑name="Calculator",  # 工具名description="计算字符长度的乘积",  # 工具信息args_schema=CalculatorInput,  # 工具接受参数信息return_direct=True,  # 直接作为工具的输出返回给调用者handle_tool_error=True,  # 报错了继续执行，不会吧那些报错行抛出，也可以自定义函数处理，handle_tool_error=函数名
)class SearchInput(BaseModel):query: str = Field(description="should be a search query")@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:"""Look up things online."""return "你好啊"tools = [search, calculator]# binding tools with llm
# 参考 https://zhuanlan.zhihu.com/p/718332876
print("test2: 使用bind_tools，将工具绑定到llm上来创建agent")
llm_with_tools = llm.bind_tools([search, calculator]
)
convert_result = llm_with_tools.invoke("`asd`的字符串长度乘以`as`的字符串长度是多少？langchain是什么？")
print(f"tool_calls = {convert_result.tool_calls}")
## 根据LLM给的参数去调用工具
# 反射获取工具
func_tool = [tool for tool in tools if tool.name == convert_result.tool_calls[0]["name"]][0]
final_result = func_tool.invoke(convert_result.tool_calls[0]["args"])
print(f"result:{final_result}")

输出结果如下：

test2: 使用bind_tools，将工具绑定到llm上来创建agent
tool_calls = [{'name': 'Calculator', 'args': {'a': '3', 'b': '2'}, 'id': '0195e80f67a9f305af774cdd63c326ff', 'type': 'tool_call'}]
result:1

d）PlanAndExecute框架：计划与执行

使用来@tool()或者StructuredTool.from_function来声明Tool；

使用load_chat_planner创建计划者，使用load_agent_executor创建执行者

使用PlanAndExecute()来创建Agent

class CalculatorInput(BaseModel):s: str = Field(description="输入字符串")def multiply(s: str) -> int:"""Multiply two numbers."""return len(s)calculator = StructuredTool.from_function(func=multiply,  # 工具具体逻辑name="Calculator",  # 工具名description="计算字符长度",  # 工具信息args_schema=CalculatorInput,  # 工具接受参数信息return_direct=True,  # 直接作为工具的输出返回给调用者handle_tool_error=True,  # 报错了继续执行，不会吧那些报错行抛出，也可以自定义函数处理，handle_tool_error=函数名
)class SearchInput(BaseModel):query: str = Field(description="should be a search query")@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:"""Look up things online."""return "你好啊"tools = [search, calculator]# https://smith.langchain.com/hub/hwchase17/react?organizationId=c4887cc4-1275-5361-82f2-b22aee75bad1
prompt = hub.pull("hwchase17/react")# 修改提示词
from langchain.prompts import PromptTemplateprompt_template = """
Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}
"""
prompt = PromptTemplate.from_template(prompt_template)
agent = create_react_agent(llm, tools, prompt)
# 查看提示词
prompt_template = agent.get_prompts()[0]# 设置计划者和执行者
from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_plannerprint("2. 设置计划者")
planner = load_chat_planner(llm)
print("3. 设置执行者")
executor = load_agent_executor(llm, tools, verbose=True)
# 初始化Plan-and-Execute Agent
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)# 运行Agent解决问题
# 有些时候会报错: return self.steps[-1][1].response, IndexError: list index out of range
agent.run("`阿萨德防守打法`有多少个字符？langchain是什么东西？前面的问题都使用中文回答，并汇总一起给我答案")
# result = agent.run("`阿萨德防守打法`有多少个字符？langchain是什么东西？前面的问题都使用中文回答，并汇总一起给我答案")
# print(f"4. 执行结果: {result}")# 运行代理解决实际问题
# 有些时候会报错: return self.steps[-1][1].response, IndexError: list index out of range
# agent.run("在纽约，100美元能买几束玫瑰？")

输出结果如下：

> Entering new AgentExecutor chain...
为了回答用户的问题，我将使用Calculator工具来计算字符串的长度。Action:
```json
{"action": "Calculator","action_input": "阿萨德防守tha"
}
Observation: 8
...

3）主流程：

施工ing…

6、回调（Callbacks）开发模板：监控任务执行过程，处理日志/异常

1）流程梳理：

施工ing…

2）代码模板：

施工ing…

3）主流程：