引言
在 2024 年的 AI 开发浪潮中，开发者往往面临两个阶段的挑战：
- 入门阶段： 如何把大模型（LLM）接入应用？（使用 LangChain）
- 进阶阶段： 如何构建能够自我纠错、多轮思考的智能体（Agent）？（使用 LangGraph）
本文将通过 6 个循序渐进的 Python 代码案例，带你从最基础的“线性链”一步步进阶到复杂的“状态机图”。
第一部分：LangChain —— 线性流水线的艺术
核心理念： Chain（链）。输入 -> 步骤A -> 步骤B -> 输出。 适用场景： 一次性问答、文档总结、简单数据处理。
案例 1：Hello World (LCEL 基础)
这是 LangChain 最经典的原子结构：Prompt + Model + Parser。 使用 LCEL (LangChain Expression Language) 的管道符 |，我们可以像搭积木一样组装逻辑。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 1. 定义组件
prompt = ChatPromptTemplate.from_template("用幽默的语气解释：什么是 {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser() # 把复杂的对象转成纯字符串
# 2. 组装链条
chain = prompt | model | parser
# 3. 运行
print(chain.invoke({"topic": "量子力学"}))💡 核心点：
|符号让数据像水流一样流过每个组件，无需写繁琐的函数调用。
案例 2：顺序链 (Sequential Chain)
当任务需要分步完成时，上一步的输出就是下一步的输入。 场景： 给产品取名 -> 根据名字写广告语。
from langchain_core.runnables import RunnablePassthrough
# 第一步：取名
name_chain = (
ChatPromptTemplate.from_template("为生产 {product} 的公司取一个酷名字")
| ChatOpenAI()
| StrOutputParser()
)
# 第二步：写标语 (接收上一步生成的 company_name)
slogan_chain = (
ChatPromptTemplate.from_template("为 {company_name} 写一句广告语")
| ChatOpenAI()
| StrOutputParser()
)
# 串联：输入 -> 算出名字 -> 传给写标语链
full_chain = {"company_name": name_chain} | slogan_chain
print(full_chain.invoke({"product": "彩虹袜子"}))💡 核心点： LangChain 自动处理了变量的传递，将复杂的逻辑拆解为简单的积木。
案例 3：最小 RAG (检索增强生成)
这是目前最主流的应用。让 AI “外挂”一个私有知识库。
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import CharacterTextSplitter
# 1. 模拟私有数据
text = "云朵公司规定：周五下午提供无限量奶茶，禁止在办公室吃螺蛳粉。"
vectorstore = FAISS.from_documents(
CharacterTextSplitter().create_documents([text]),
OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
# 2. 构建 RAG 链
template = "基于已知信息回答：{context}。问题：{question}"
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| ChatPromptTemplate.from_template(template)
| ChatOpenAI()
| StrOutputParser()
)
print(rag_chain.invoke("在公司吃螺蛳粉会怎样？"))💡 核心点：
Retriever充当了搜索引擎的角色，让 AI 能够回答训练数据之外的问题。
第二部分：LangGraph —— 循环与状态机的威力
核心理念： Graph（图）。引入了循环（Cycle）、状态（State）和分支（Branching）。 适用场景： 复杂 Agent、代码生成、需要人工确认的流程。
案例 4：基础循环 (The Loop) —— 自动纠错
LangGraph 允许 AI 检查自己的输出，如果不满意就重做。 场景： 写手生成内容 <-> 编辑检查内容。
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List
# 1. 定义状态 (共享内存)
class State(TypedDict):
messages: List[str]
iteration: int
# 2. 定义节点
def writer(state):
# 模拟生成
return {"messages": ["生成的草稿"], "iteration": state["iteration"] + 1}
def editor(state):
# 模拟检查：如果次数少于3次，就打回重写
if state["iteration"] < 3:
return "continue" # 触发条件边
return "end"
# 3. 建图
workflow = StateGraph(State)
workflow.add_node("writer", writer)
workflow.add_node("editor", editor) # 这里简化为一个逻辑判断节点
workflow.add_edge(START, "writer")
workflow.add_conditional_edges(
"writer",
editor, # 使用 editor 函数作为路由逻辑
{"continue": "writer", "end": END} # 映射关系
)
app = workflow.compile()
app.invoke({"messages": [], "iteration": 0})💡 核心点：
writer指向editor，而editor又指回writer，形成了一个闭环，直到条件满足才退出。
案例 5：条件路由 (Router) —— 智能分诊
摒弃死板的 if-else，让 LLM 根据语义决定任务交给谁处理。
def classify_intent(state):
question = state["question"]
if "计算" in question: return "math"
return "general"
workflow = StateGraph(dict) # 简单状态
workflow.add_node("math_node", lambda x: {"answer": "这是数学题"})
workflow.add_node("general_node", lambda x: {"answer": "这是普通聊天"})
# 条件边：从起点直接分流
workflow.add_conditional_edges(
START,
classify_intent,
{"math": "math_node", "general": "general_node"}
)
app = workflow.compile()
print(app.invoke({"question": "计算 1+1"}))💡 核心点： 这种架构让应用具备了“大脑”，可以根据用户意图动态选择工具。
案例 6：人机协同 (Human-in-the-loop)
这是企业级应用的神器。在关键步骤暂停，等待人类批准。
from langgraph.checkpoint.memory import MemorySaver
# 1. 建图
workflow = StateGraph(dict)
workflow.add_node("write_email", lambda x: {"draft": "邮件草稿..."})
workflow.add_node("send_email", lambda x: print("🚀 发送成功"))
workflow.add_edge(START, "write_email")
workflow.add_edge("write_email", "send_email")
# 2. 关键配置：内存 + 中断
checkpointer = MemorySaver()
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["send_email"] # 在发送前暂停！
)
# 3. 运行阶段 1
thread = {"configurable": {"thread_id": "1"}}
app.invoke({"topic": "汇报"}, config=thread)
print("--- 系统已暂停，等待人工审批 ---")
# 4. 运行阶段 2 (人类批准后)
# 传入 None 表示继续之前的状态
app.invoke(None, config=thread) 💡 核心点：
interrupt_before实现了类似于“断点调试”的功能，配合checkpointer实现了长周期的任务管理。
总结
- 如果你做的是“直来直去”的任务（如简单的文档问答），LangChain 是最高效的选择，代码简洁，开箱即用。
- 如果你做的是“需要动脑子”的任务（如写代码、长流程规划、需要审批的操作），LangGraph 是必经之路，它提供了构建智能体所需的循环与记忆能力。
希望这 6 个案例能成为你 AI 开发路上的“脚手架”，助你快速构建下一代智能应用！
版权属于：soarli
本文链接：https://blog.soarli.top/archives/784.html
转载时须注明出处及本声明。
