第11章从死脑筋到会拐弯

第10章我们挑好了框架。这一章，正式动手——把LangChain的"链"和LangGraph的"图"吃透。从一条直线串到底的简单Agent，到能拐弯、能循环、能记状态的复杂Agent，你会看到Agent从"死脑筋"变成"会拐弯"的全过程。

11.1 把Agent的"流水线"搭起来

11.1.1 LCEL：一个管道符串起一切

LangChain 的核心是 LCEL（LangChain Expression Language）。说白了就一个操作符：|。

python

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 三块积木
prompt = ChatPromptTemplate.from_template("讲个关于{topic}的笑话")
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 一个管道符串起来
chain = prompt | model | parser

# 调用
result = chain.invoke({"topic": "程序员"})
print(result)

prompt | model | parser 的意思是：数据从 prompt 流出，经过 model 处理，再经过 parser 解析，最后输出。

11.1.2 写一条能查天气的链

单步链太简单。我们来写一条真正的Agent链——用户问天气，Agent去查。

python

from langchain.tools import tool
from langchain_openai import ChatOpenAI

# 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    weather_db = {"北京": "晴，25度", "上海": "多云，28度"}
    return weather_db.get(city, "未找到")

# 绑定工具到模型
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools([get_weather])

# 调用
from langchain_core.messages import HumanMessage
response = llm_with_tools.invoke([
    HumanMessage(content="北京天气怎么样？")
])

# 看LLM决定调什么工具
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}}]

关键点：bind_tools 之后，LLM 不再直接回答，而是返回"该调什么工具"的指令。Agent拿到这个指令，执行工具，再把结果喂回LLM。

11.2 让Agent"说话有条理"

11.2.1 Prompt模板：不是你随便写的字符串

在实际项目中，Prompt很少是写死的字符串。用户每次输入不同，Prompt要动态拼接。

python

from langchain.prompts import ChatPromptTemplate

# 定义一个模板
template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}。回答风格：{style}。"),
    ("user", "{question}")
])

# 不同场景灌不同参数
customer_service = template.invoke({
    "role": "客服助手",
    "style": "亲切友善",
    "question": "我的订单什么时候到？"
})

technical_writer = template.invoke({
    "role": "技术文档撰写者",
    "style": "严谨专业",
    "question": "解释一下什么是RAG。"
})

模板的好处：同一个Prompt结构，换不同的参数就能适配不同场景。Agent系统里大量用这种模式。

11.2.2 输出解析器：别让LLM乱说话

LLM的输出是自由文本，但Agent需要结构化数据。输出解析器把LLM的"人话"转成程序能用的格式。

python

from langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

# 定义想要的输出结构
class WeatherReport(BaseModel):
    city: str = Field(description="城市名")
    temperature: int = Field(description="温度")
    condition: str = Field(description="天气状况")

parser = PydanticOutputParser(pydantic_object=WeatherReport)

# 把要求的格式写进Prompt
prompt = ChatPromptTemplate.from_template("""
查询{city}的天气。请用JSON格式返回，包含city、temperature、condition字段。
{format_instructions}
""")

chain = prompt | llm | parser
result = chain.invoke({
    "city": "北京",
    "format_instructions": parser.get_format_instructions()
})

# result 是一个 WeatherReport 对象，可以直接用 result.city, result.temperature

11.3 多轮对话怎么存？

11.3.1 问题：LLM天生"健忘"

每次调用 llm.invoke() 都是一个独立的请求。LLM不知道上一轮说了什么。

python

# 第一轮
llm.invoke([HumanMessage(content="北京天气怎么样？")])
# "北京今天晴，25度"

# 第二轮——LLM不知道上一轮问了什么
llm.invoke([HumanMessage(content="那明天呢？")])
# "请问您说的是哪个城市的明天？"（忘了！）

11.3.2 解决方案：把历史消息存起来

每次对话，把之前的消息拼进新请求里。

python

from langchain_core.messages import HumanMessage, AIMessage

# 维护一个消息列表
messages = []

# 第一轮
messages.append(HumanMessage(content="北京天气怎么样？"))
response = llm.invoke(messages)
messages.append(response)  # 把LLM的回答也存进去
# "北京今天晴，25度"

# 第二轮——带上历史
messages.append(HumanMessage(content="那明天呢？"))
response = llm.invoke(messages)
# "北京明天多云，22度"（知道在问北京了！）

11.3.3 消息管理的高级玩法

实际项目中不能无限存消息——token会爆炸。需要"修剪"：

python

from langchain_core.messages import trim_messages

# 只保留最近10轮对话
trimmed = trim_messages(
    messages,
    max_tokens=2000,       # 最多2000 token
    strategy="last",       # 保留最新的
    token_counter=llm,     # 用LLM的token计数
    include_system=True,   # system消息始终保留
)

11.4 当"一条直线"不够用的时候

11.4.1 LCEL的极限

LCEL的链是单向的：数据从左流到右，不会回头。但Agent经常需要"回头"——工具返回的结果不满意，重新查；规划错了，回到上一步。

这个时候，你需要LangGraph。

11.4.2 StateGraph：把Agent变成状态机

LangGraph的核心是 StateGraph——把Agent的行为定义成节点（Node）和边（Edge）。

python

from langgraph.graph import StateGraph, END
from typing import TypedDict

# 1. 定义状态
class AgentState(TypedDict):
    messages: list
    next_step: str

# 2. 定义节点（每个节点是一个函数）
def call_model(state):
    """调用LLM"""
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def call_tool(state):
    """执行工具"""
    last_msg = state["messages"][-1]
    tool_result = execute_tool(last_msg.tool_calls[0])
    return {"messages": state["messages"] + [tool_result]}

def should_continue(state):
    """判断：继续调工具，还是结束？"""
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "call_tool"
    return "end"

# 3. 构建图
graph = StateGraph(AgentState)
graph.add_node("call_model", call_model)
graph.add_node("call_tool", call_tool)
graph.add_edge("call_model", "call_tool")        # 模型调用后→工具调用
graph.add_conditional_edges("call_tool", should_continue, {
    "call_tool": "call_model",  # 继续调工具→回到模型
    "end": END                   # 结束
})
graph.set_entry_point("call_model")

app = graph.compile()

11.4.3 条件边：让Agent自己决定往哪走

add_conditional_edges 是LangGraph的灵魂。它不是写死的分支，而是运行时动态决定下一步走哪个节点。

python

def router(state):
    """根据LLM的判断，决定下一步"""
    last_msg = state["messages"][-1]
    
    if "天气" in str(last_msg):
        return "weather_tool"
    elif "新闻" in str(last_msg):
        return "news_tool"
    elif "计算" in str(last_msg):
        return "calculator_tool"
    else:
        return "fallback"

graph.add_conditional_edges("call_model", router, {
    "weather_tool": "call_weather",
    "news_tool": "call_news",
    "calculator_tool": "call_calculator",
    "fallback": "no_tool_response"
})

11.5 让Agent"睡一觉起来还能接着干"

11.5.1 检查点：把状态存盘

默认情况下，LangGraph的Agent每次会话结束，状态就丢了。**检查点（Checkpoint）**让Agent能把状态保存下来，下次继续。

python

from langgraph.checkpoint.memory import MemorySaver

# 创建检查点存储器
memory = MemorySaver()

# 编译时传入检查点
app = graph.compile(checkpointer=memory)

# 第一次对话
config = {"configurable": {"thread_id": "user-123"}}
app.invoke({"messages": [HumanMessage(content="北京天气？")]}, config)

# 第二次对话——带着同样的 thread_id
app.invoke({"messages": [HumanMessage(content="那明天呢？")]}, config)
# Agent知道你在问"北京明天"，因为检查点保存了上一轮的上下文

11.5.2 线程隔离：不同用户互不干扰

thread_id 就是"用户ID"。不同的thread_id，不同的检查点，互不影响。

python

# 用户A的对话
app.invoke({"messages": [HumanMessage(content="你好")]}, 
           {"configurable": {"thread_id": "user-A"}})

# 用户B的对话——完全独立
app.invoke({"messages": [HumanMessage(content="你好")]}, 
           {"configurable": {"thread_id": "user-B"}})

11.5.3 时间旅行：回到之前的某个状态

检查点还支持"时间旅行"——你可以回到对话的任意一个历史节点，从那里重新开始。

python

# 查看历史状态
history = list(app.get_state_history(config))

# 回到第3个状态
app.update_state(config, history[3].values)

# 从这里继续对话
app.invoke({"messages": [HumanMessage(content="换个思路重新来？")]}, config)

11.6 LangGraph 的条件路由与 ReAct

第 11.4 节讲了 StateGraph 的基础骨架。这一节深入两个核心机制：条件路由的更多实战写法和 ReAct 循环模式。

11.6.1 条件路由：根据多维度做决策

基础的条件路由只判断一个维度的关键词。实际场景中，路由函数经常要综合多个条件：

python

def complex_router(state: AgentState) -> str:
    msg = state["messages"][-1].content
    urgency = state.get("urgency_level", 0)
    is_vip = state.get("is_vip", False)

    # VIP + 高紧急度 → 人工客服
    if is_vip and urgency > 8:
        return "human_agent"
    # 高紧急度 → 快速通道
    elif urgency > 7:
        return "fast_track"
    # VIP → VIP 通道
    elif is_vip:
        return "vip_queue"
    # 普通 → 标准流程
    else:
        return "standard_flow"

add_conditional_edges 拿到路由函数的返回值，查找到对应的目标节点。只要函数返回字符串，条件可以任意复杂。

11.6.2 ReAct 循环：思考 → 行动 → 观察 → 再思考

ReAct（Reasoning + Acting）是 Agent 推理的经典模式。LangGraph 用三个节点 + 一个循环边原生支持：

python

from langgraph.graph import StateGraph, START, END

graph = StateGraph(ReActState)

graph.add_node("reasoner", reason_node)      # 推理
graph.add_node("act", act_node)              # 执行工具
graph.add_node("observe", observe_node)       # 观察结果

graph.add_edge(START, "reasoner")
graph.add_edge("reasoner", "act")
graph.add_edge("act", "observe")
# 关键：观察完回到推理，形成循环
graph.add_edge("observe", "reasoner")

# 终止条件
def should_continue(state: ReActState) -> Literal["reasoner", "END"]:
    if state["should_stop"]:
        return "END"
    return "reasoner"

graph.add_conditional_edges("observe", should_continue, {
    "reasoner": "reasoner",
    "END": END
})

执行路径：Reason（推理/规划）→ Act（执行行动）→ Observe（观察结果）→ 判断是否结束，否则回到 Reason。

11.6.3 防止无限循环：step 计数器

ReAct 循环最大的隐患是"无限循环"——LLM 反复调用工具但得不到满意结果。解决方案：在 State 中加一个 step 字段，每次循环自增，超限就强制退出。

python

class SafeState(TypedDict):
    messages: Annotated[list, add]
    step: int

def check_step_limit(state: SafeState) -> Literal["continue", "END"]:
    if state["step"] >= 10:  # 最多循环 10 次
        return "END"
    return "continue"

def increment_step(state: SafeState) -> dict:
    return {"step": state["step"] + 1}

面试常考题："多 Agent 如何避免循环调用？"——答案就是这个 step 计数器。简单，但实用。

11.7 本章小结

这一章我们从"直线"走到了"回路"：

LCEL：一个 | 串起Prompt、Model、Parser。简单场景一把梭。
Prompt模板和解析器：模板让Prompt复用，解析器让LLM输出结构化。
多轮对话记忆：把历史消息拼进去，再用 trim_messages 控制token。
StateGraph：节点+边+条件跳转。Agent从"死脑筋"变成"会拐弯"。
检查点：存盘、读取、隔离、时间旅行。让Agent有"持久记忆"。

LangChain让Agent跑起来，LangGraph让Agent跑得聪明。

✅ 知识点检查

学完这一章，试试回答这几个问题：

[ ] LCEL的 | 操作符做了什么？链的流向是怎样的？
[ ] Prompt模板和输出解析器各解决了什么问题？
[ ] StateGraph和LCEL链的核心区别是什么？
[ ] 检查点的三个作用是什么？

📚 延伸阅读

LangChain LCEL文档：https://python.langchain.com/docs/expression_language/
LangGraph StateGraph文档：https://langchain-ai.github.io/langgraph/concepts/low_level/
本书配套源码：关注公众号「图解AI系列」免费领取

🎯 下一章预告

第12章，Agent上线了，但它靠谱吗——

"上了线，就放心了？安全漏洞、幻觉问题、评估指标……先别急着庆祝。"

第11章 从死脑筋到会拐弯 ​

11.1 把Agent的"流水线"搭起来 ​

11.1.1 LCEL：一个管道符串起一切 ​

11.1.2 写一条能查天气的链 ​

11.2 让Agent"说话有条理" ​

11.2.1 Prompt模板：不是你随便写的字符串 ​

11.2.2 输出解析器：别让LLM乱说话 ​

11.3 多轮对话怎么存？ ​

11.3.1 问题：LLM天生"健忘" ​

11.3.2 解决方案：把历史消息存起来 ​

11.3.3 消息管理的高级玩法 ​

11.4 当"一条直线"不够用的时候 ​

11.4.1 LCEL的极限 ​

11.4.2 StateGraph：把Agent变成状态机 ​

11.4.3 条件边：让Agent自己决定往哪走 ​

11.5 让Agent"睡一觉起来还能接着干" ​

11.5.1 检查点：把状态存盘 ​

11.5.2 线程隔离：不同用户互不干扰 ​

11.5.3 时间旅行：回到之前的某个状态 ​

11.6 LangGraph 的条件路由与 ReAct ​

11.6.1 条件路由：根据多维度做决策 ​

11.6.2 ReAct 循环：思考 → 行动 → 观察 → 再思考 ​

11.6.3 防止无限循环：step 计数器 ​

11.7 本章小结 ​

✅ 知识点检查 ​

📚 延伸阅读 ​

🎯 下一章预告 ​

喜欢这篇内容？关注我，别错过后续更新

第11章从死脑筋到会拐弯

11.1 把Agent的"流水线"搭起来

11.1.1 LCEL：一个管道符串起一切

11.1.2 写一条能查天气的链

11.2 让Agent"说话有条理"

11.2.1 Prompt模板：不是你随便写的字符串

11.2.2 输出解析器：别让LLM乱说话

11.3 多轮对话怎么存？

11.3.1 问题：LLM天生"健忘"

11.3.2 解决方案：把历史消息存起来

11.3.3 消息管理的高级玩法

11.4 当"一条直线"不够用的时候

11.4.1 LCEL的极限

11.4.2 StateGraph：把Agent变成状态机

11.4.3 条件边：让Agent自己决定往哪走

11.5 让Agent"睡一觉起来还能接着干"

11.5.1 检查点：把状态存盘

11.5.2 线程隔离：不同用户互不干扰

11.5.3 时间旅行：回到之前的某个状态

11.6 LangGraph 的条件路由与 ReAct

11.6.1 条件路由：根据多维度做决策

11.6.2 ReAct 循环：思考 → 行动 → 观察 → 再思考

11.6.3 防止无限循环：step 计数器

11.7 本章小结

✅ 知识点检查

📚 延伸阅读

🎯 下一章预告