在对大语言模型应用开发框架的探索中,LangChain 和 LangGraph 是两个核心且关联紧密的名字。为了帮助你快速建立整体认知,下面这个表格清晰地展示了它们的核心定位、关键特性和典型适用场景。
🔄 核心概念解析
要理解如何使用它们,需要掌握一些核心概念:
LangChain的核心:
组件:提供了文档加载器、文本分割器、各种向量数据库接口、提示模板等大量可即插即用的模块。
链:这是LangChain的灵魂,通过LCEL将多个组件(如提示模板、模型、输出解析器)连接成一个可执行的工作流。
代理:赋予模型使用工具(如计算器、搜索引擎)的能力,让其可以主动调用外部资源完成任务。
记忆:用于在对话或多次调用间保持状态。
LangGraph的核心:
图:将应用流程定义为由节点和边构成的有向图。节点代表一个操作步骤,边定义了步骤间的流转路径。
状态:一个共享的数据结构,在图中的节点间传递和修改,这是实现复杂、有状态交互的基石。
条件边:根据当前状态的值,动态决定下一个要执行的节点,从而实现
if-else逻辑。
🛠️ 如何使用与适用场景
根据你的项目需求,可以参考以下指南进行选型:
选择 LangChain 当你的项目是:
简单的线性任务:如文本翻译、内容摘要、单次问答。这些任务输入明确,步骤清晰,不需要复杂的循环或状态跟踪。
需要快速原型验证:利用其丰富的组件,可以快速搭建一个可用的演示系统。
作为基础组件库:即使在LangGraph项目中,也常使用LangChain提供的模型接口、文档加载器等组件。
选择 LangGraph 当你的项目是:
复杂的、多步骤的智能体:例如,一个能自动分析问题、联网搜索、撰写报告并检查质量的系统。这类任务需要多次尝试和决策。
复杂的多轮对话:如客服系统、教育辅导AI,需要深刻理解上下文,并基于历史对话进行回复。
多智能体协作系统:需要多个AI智能体扮演不同角色(如分析师、工程师、评论员)共同完成一项任务。
💻 完整示例:构建一个报告审阅Agent
下面我们用一个具体的例子来展示两者的区别。假设我们要构建一个报告审阅Agent,其逻辑是:生成营销文案 -> 自动进行合规检查 -> 如果检查不通过,则重新生成直至通过。
示例1:使用 LangChain 实现(外部循环控制)
这种方式下,循环的重试逻辑由开发者在应用层手动控制,工作流本身是线性的。
from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate
# 1. 定义模型和提示
llm = ChatOpenAI(model="gpt-4")
prompt_template = ChatPromptTemplate.from_template("为产品 {product} 生成一段营销文案。")
compliance_prompt = ChatPromptTemplate.from_template("检查以下文案是否合规:{draft}")
# 2. 创建链
copy_chain = LLMChain(llm=llm, prompt=prompt_template)
check_chain = LLMChain(llm=llm, prompt=compliance_prompt)
# 3. 外部循环控制逻辑
product = "新款防晒霜"
content_approved = False
max_retries = 3
retry_count = 0
while not content_approved and retry_count < max_retries:
# 生成初稿
draft = copy_chain.run(product=product)
print(f"生成的文案:{draft}")
# 合规检查
feedback = check_chain.run(draft=draft)
print(f"合规检查结果:{feedback}")
# 简单判断是否包含违规词(实际逻辑更复杂)
if "违规" not in feedback:
content_approved = True
print("文案通过!")
else:
retry_count += 1
print(f"文案未通过,尝试重新生成... ({retry_count}/{max_retries})")说明:这个示例的循环是通过外部的while循环实现的,流程控制与业务逻辑耦合在一起,扩展复杂步骤时容易变得难以维护。
示例2:使用 LangGraph 实现(内置工作流)
这种方式将循环和判断逻辑内置在图的工作流中,结构更清晰,更易于扩展和调试。
from langgraph.graph import StateGraph, END
from typing import TypedDict
from langchain.chat_models import ChatOpenAI
from langchain.schema import BaseMessage, HumanMessage
# 1. 定义状态结构(共享的数据结构)
class AgentState(TypedDict):
product: str
draft: str
feedback: str
legal_status: str # 'PENDING', 'APPROVED', 'REJECTED'
attempts: int
# 2. 定义各个节点(功能步骤)
llm = ChatOpenAI(model="gpt-4")
def generate_draft(state: AgentState) -> AgentState:
"""节点A:生成文案草稿"""
prompt = f"为产品 {state['product']} 生成一段营销文案。"
message = HumanMessage(content=prompt)
response = llm([message])
new_draft = response.content
print(f"【生成节点】生成的文案:{new_draft}")
return {**state, "draft": new_draft, "attempts": state.get("attempts", 0) + 1}
def legal_review(state: AgentState) -> AgentState:
"""节点B:进行合规审查"""
prompt = f"检查以下文案是否合规,仅回复'APPROVED'或'REJECTED'及原因:{state['draft']}"
message = HumanMessage(content=prompt)
response = llm([message])
feedback = response.content
print(f"【审查节点】审查意见:{feedback}")
if "APPROVED" in feedback:
legal_status = "APPROVED"
else:
legal_status = "REJECTED"
return {**state, "feedback": feedback, "legal_status": legal_status}
def human_approval(state: AgentState) -> AgentState:
"""节点C(终结点):人工确认"""
print("【人工节点】文案已通过自动审查,待人工最终确认。")
# 此处可集成真实的人工审核接口
return {**state, "legal_status": "APPROVED"}
# 3. 构建图工作流
builder = StateGraph(AgentState)
# 添加节点
builder.add_node("generate", generate_draft)
builder.add_node("legal_review", legal_review)
builder.add_node("human_approval", human_approval)
# 设置入口点
builder.set_entry_point("generate")
# 连接节点:生成 -> 审查
builder.add_edge("generate", "legal_review")
# 添加条件边:根据审查结果决定下一步
def route_after_review(state: AgentState) -> str:
if state["legal_status"] == "APPROVED":
return "proceed_to_human" # 通过,走向人工确认
elif state["attempts"] >= 3: # 重试超过3次,终止
return "end"
else:
return "retry" # 不通过,重新生成
builder.add_conditional_edges(
"legal_review",
route_after_review,
{
"proceed_to_human": "human_approval",
"retry": "generate",
"end": END
}
)
# 连接节点:人工确认 -> 结束
builder.add_edge("human_approval", END)
# 编译图
graph = builder.compile()
# 4. 执行工作流
initial_state = {"product": "全新智能手表"}
result = graph.invoke(initial_state, config={"recursion_limit": 5}) # 限制递归深度
print("\n=== 最终结果 ===")
print(f"产品:{result['product']}")
print(f"通过的文案:{result['draft']}")示例解析:
状态:
AgentState是所有节点共享和操作的“白板”。节点:每个节点(
generate,legal_review)负责一个具体任务。条件边:
route_after_review函数是大脑,根据legal_status和attempts决定下一步是重试、转人工还是结束。优势:工作流的逻辑被封装在图中,可视化强,易于增加新节点(如增加一个
optimize优化节点),错误恢复能力内建。
💎 如何选择与总结
简单来说,选择取决于你的任务复杂度和对“状态”的需求。
LangChain 是你的瑞士军刀,适合定义明确、一步到位的任务。它通过丰富的组件和链式调用,让你能快速搭建应用。
LangGraph 是你的项目总指挥,适合需要动态规划、多步执行并能从错误中自我恢复的复杂场景。它通过图结构和状态管理,为你构建强大的自主智能体提供了坚实基础。
在实际项目中,它们常常协同工作:利用LangChain强大的组件库作为“武器”,再由LangGraph这位“指挥官”来编排复杂的作战计划。
希望这个详细的对比和示例能帮助你做出清晰的选择!如果你有更具体的应用场景想法,我们可以继续深入探讨。
评论区