浅谈LangChain\LangGraph(下)
前言：
之前已经在本地跑通了LangChain的Demo项目了，接下来要改造成LangGraph，首先分析一下langGraph的几大优势点：
我的个人总结：
1、从线性管道到状态机节点（类似从二维升级到了三维，单线程变成多线程）
2、控制流变成循环、分支，比原先的单向高级了
3、容错性变强、支持回滚
4、适用的场景变多了(简单的RAG单次调用变成了复杂工作流多Agent协同工作)
（官方对比LangGraph和LangChain的回答网上很多，此处省略了）
最主要还有一点原因：
传统的 AgentExecutor 在 LangChain 官方已经被逐步废弃，LangGraph 已经成为 LangChain 生态中构建复杂 Agent 的官方标准架构。
1、改造准备：
节点网络：
节点网络对应图架构里的“点”和“边”：
# 1. 定义节点
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
# 2. 定义边、构建网络走向
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tools", END])
builder.add_edge("tools", "agent")控制流循环：
LangGraph 的有环图（相比较于LangChain的A -> B -> C）：
builder.add_conditional_edges("agent", should_continue, ["tools", END]) #
builder.add_edge("tools", "agent") 状态机与状态传递：
状态是图在运行过程中共享的“全局单一事实源”：
builder = StateGraph(MessagesState)
async def agent_node(state: MessagesState, config: RunnableConfig) -> Dict[str, Any]:
response = await self.llm_with_tools.ainvoke(state["messages"], config=config)[cite: 3]
return {"messages": [response]}自我纠错机制：
反馈回路： 当 tools 节点执行失败（例如 API 返回 404 或报错）时，ToolNode 会将错误信息包装成 ToolMessage 写入 State，并沿着 tools -> agent 边传回给 agent 节点[cite: 3]。配合在系统提示词里写的规则:
"1. 如果调用的工具返回了错误或‘未收录’等提示，必须如实告知用户无法提供该信息，绝对严禁凭空捏造数据！"[cite: 3]持久化：
应用层持久化：
saved_messages: List[BaseMessage] = history_store.messages[cite: 3]
new_messages_to_save = all_graph_messages[initial_msg_count:][cite: 3]
history_store.add_messages([current_human_msg] + new_messages_to_save)[cite: 3]LangGraph 原生持久化：
graph = builder.compile(checkpointer=RedisSaver(redis_conn))2、代码升级：
完成上面的改造后，贴出完整的服务层升级代码：
（为了不用动其他层的代码，此处类命名依然为AgentService，有点类似AOP的思想，实际是想省点事）
服务层代码
import json
import logging
from typing import AsyncGenerator, List, Dict, Any, Literal
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, BaseMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
from app.core.llm import get_llm
from app.tools import ALL_TOOLS, TOOLS_MAP
logger = logging.getLogger(__name__)
class AgentService:
def __init__(self, redis_url: str = "redis://localhost:6379/0"):
self.llm = get_llm()
self.tools_map = TOOLS_MAP
self.redis_url = redis_url
self.llm_with_tools = self.llm.bind_tools(ALL_TOOLS)
self.graph = self._build_graph()
def _get_history(self, session_id: str) -> RedisChatMessageHistory:
return RedisChatMessageHistory(
session_id=session_id,
url=self.redis_url,
ttl=86400
)
def _build_graph(self):
async def agent_node(state: MessagesState, config: RunnableConfig) -> Dict[str, Any]:
response = await self.llm_with_tools.ainvoke(state["messages"], config=config)
return {"messages": [response]}
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
messages = state["messages"]
last_message = messages[-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return "tools"
return END
tool_node = ToolNode(ALL_TOOLS)
builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tools", END])
builder.add_edge("tools", "agent")
return builder.compile()
async def get_stream_response(self, question: str, session_id: str) -> AsyncGenerator[str, None]:
try:
history_store = self._get_history(session_id)
system_prompt = SystemMessage(
content=(
"你是一个严谨且高效的智能助手。\n"
"规则要求：\n"
"1. 如果调用的工具返回了错误或‘未收录’等提示，必须如实告知用户无法提供该信息，绝对严禁凭空捏造数据！\n"
"2. 严格基于工具返回的真实结果进行回答。"
)
)
saved_messages: List[BaseMessage] = history_store.messages
current_human_msg = HumanMessage(content=question)
input_messages = [system_prompt] + saved_messages + [current_human_msg]
initial_state = {"messages": input_messages}
initial_msg_count = len(input_messages)
final_state: Dict[str, Any] = {}
async for event in self.graph.astream_events(initial_state, version="v2"):
event_type = event["event"]
# 捕获LLM逐字Token
if event_type == "on_chat_model_stream":
chunk = event["data"]["chunk"]
if chunk.content:
payload = json.dumps({"content": chunk.content}, ensure_ascii=False)
yield f"data: {payload}\n\n"
#捕获State后续保存Redis
elif event_type in ("on_chain_end", "on_graph_end"):
if "output" in event["data"] and isinstance(event["data"]["output"], dict):
if "messages" in event["data"]["output"]:
final_state = event["data"]["output"]
#持久化记录至Redis
if "messages" in final_state:
all_graph_messages = final_state["messages"]
new_messages_to_save = all_graph_messages[initial_msg_count:]
history_store.add_messages([current_human_msg] + new_messages_to_save)
except Exception as e:
#捕捉异常，避免StreamingResponse静默挂起
logger.error(f"[AgentService 运行时异常]: {str(e)}", exc_info=True)
err_payload = json.dumps({"error": f"服务运行异常: {str(e)}"}, ensure_ascii=False)
yield f"data: {err_payload}\n\n"3、技术总结：
这里有个很坑的地方：
LangGraph默认更偏向整块吐出，LangChain的设计视角是模型，LangGraph的设计视角是状态机，核心是节点变迁和状态更新，所以LangGraph的五种流式模式（分别是values、updates、messages、custom、debug），这里都没有使用，这里使用的是astream_events。
这是一个更底层的事件总线API，至于为啥要用这个，调试流式逐字打印，调试了很多都达不到满意的效果，我想要LangChain的逐字输出，最终使用了这个方法。astream_events不依赖任何单一的stream_model，而是将整个图在运行过程中产生的所有细粒度事件全部广播出来。
以上就是本次简单练习FastApi+SSE+LangChain到LangGraph的浅谈总结，最坑的其实还是这个提到的逐字流式输出，只有尝试过，才知道，调了数十次，要么前端无响应空白，要么就整块往外吐，研究完底层文档后，找到了最优解，值得花了大半天时间攻克这个难题。