微信扫码
添加专属顾问
我要投稿
LangGraph凭借其开放性和灵活性,成为企业级AI智能体开发的首选框架,助你摆脱厂商锁定,打造定制化解决方案。 核心内容: 1. LangGraph与OpenAI Assistant API的灵活性对比 2. LangGraph的底层架构与LangChain的紧密关系 3. 开发实践中的优势与挑战分析
接入方式上,LangGraph也很人性化。你可以用传统的OpenAI API方式集成,也可以用Ollama或vLLM这样的推理加速库,让整个过程更高效便捷。在AI Agent的构建范式上,LangGraph不只提供预设的ReAct机制,还允许自定义Planning策略,适应不同场景的需求。
从这些角度看,LangGraph的自主性和开放性确实碾压Assistant API。但话说回来,这种灵活性也带来了挑战。LangGraph要求开发者做更多自主工作,底层架构复杂,学习曲线陡峭。我自己实践了好几轮,才勉强上手。即使优化再多,用LangGraph建的Agent效果也不一定超过Assistant API几行代码就能搞定的简单应用。所以,如果你打算入坑,得做好花时间精力的准备——这不是一蹴而就的事儿。
在LangChain的体系中,创建代理(Agent)是个核心用例。它的底层架构分成三个部分:Agent、Message和Toolkits。Agent通常由大语言模型、提示词和输出解析器组成,形成一个Chain来处理输入。输入来源有用户原始输入、模型响应和历史上下文。输出则连接到工具库,根据子任务决定调用哪些工具。
有趣的是,LangChain的Agent不直接执行操作,它只负责决策。实际执行交给AgentExecutor这个"运行时"组件。结合起来,才形成完整的智能体。这套设计很巧妙,但随着AI应用越来越复杂,LangChain的线性序列(基于LCEL的DAG,有向无环图)开始显露出局限性。开发者需要更多循环、分支和复杂交互,这时候LangGraph就应运而生了。
举个简单例子来说明吧。想象一个基本的代理模型:用户输入先到"Start"节点(比如一个大模型API),它决定是否需要检索信息。如果需要,就跳到"Action"节点,用工具如Web Search或数据库查询获取数据。然后,再用LLM处理这些信息,生成响应,最后到"End"节点结束。
这里的关键是,每个圆圈是"节点"(Nodes),箭头是"边"(Edges)。LangGraph的工作流就是节点完成任务后,通过边传递消息到下一个节点。消息传递像程序的通用定义:节点处理、发送消息、下一个节点接力,循环往复。这就是LangGraph图算法的核心思想。
再看个复杂点的场景。假设起点是AI Agent,它配置了多个工具。用户问题进来,Agent判断:
不需要工具,直接自然语言响应。比如问"你好",就回"我是AI助手"。
需要工具,就输出JSON格式的函数调用,比如调用web_search查询"什么是智能体AI"。
如果有多个工具可选,LangGraph用"条件边"决定路径,像if-else语句。通过"Router"组件,基于条件(如是否需要搜索)导向不同节点:搜索路径获取文本,再用LLM处理;直接回答路径用工具格式化输出。
Router的实现有三种方式:提示工程(指导模型特定格式响应)、输出解析器(提取结构化数据)、工具调用(用模型内置功能)。
从官方定义,LangGraph是用于构建有状态、多参与者应用的库,特别适合代理和多代理工作流。它的优势包括:
循环和分支:轻松实现条件和循环
持久性:每步自动保存状态,支持暂停、恢复、错误恢复和人机交互
人机交互:中断执行,批准或编辑下一步操作
流支持:节点生成流输出,包括令牌流
与LangChain集成:无缝兼容LangChain和LangSmith,但不强制依赖
既然说了这么多理论,咱们直接上手试试吧。在开始之前,我们需要安装必要的包:
pip install langgraph langchain-openai
如果你用的是国内环境,可能需要配置代理或者使用其他大模型服务。
1. 基础设置
import osfrom typing import TypedDict, Annotatedfrom langgraph.graph import StateGraph, END, STARTfrom langgraph.graph.message import add_messagesfrom langchain_openai import ChatOpenAI# 设置API密钥os.environ["OPENAI_API_KEY"] = "你的API密钥"# 初始化模型llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
2. 定义状态
在LangGraph中,状态是整个工作流的核心,它存储了流程中的所有信息:
class WritingState(TypedDict): messages: Annotated[list, add_messages] content_type: str tone: str length: str draft: str final_output: str
这个状态定义了我们写作助手需要跟踪的所有信息。
3. 创建节点函数
节点是工作流中的每个处理步骤,我们需要定义几个关键节点:
def analyze_request(state: WritingState): """分析用户请求,确定内容类型和写作参数""" user_message = state["messages"][-1].content analysis_prompt = f""" 分析以下用户请求,提取关键信息: 用户请求:{user_message} 请以JSON格式回复,包含: - content_type: 内容类型(如:博客文章、邮件、报告等) - tone: 语调(如:正式、友好、专业等) - length: 长度要求(如:短、中、长) """ response = llm.invoke(analysis_prompt) # 这里简化处理,实际项目中你可能需要更复杂的解析逻辑 return { "content_type": "博客文章", "tone": "友好", "length": "中等" }def generate_outline(state: WritingState): """根据分析结果生成大纲""" user_message = state["messages"][-1].content outline_prompt = f""" 用户需求:{user_message} 内容类型:{state['content_type']} 语调:{state['tone']} 长度:{state['length']} 请为此需求生成一个详细的大纲。 """ response = llm.invoke(outline_prompt) return {"draft": response.content}def write_content(state: WritingState): """基于大纲生成最终内容""" user_message = state["messages"][-1].content writing_prompt = f""" 用户需求:{user_message} 内容大纲:{state['draft']} 请根据大纲创作完整的{state['content_type']}, 语调要{state['tone']},长度{state['length']}。 """ response = llm.invoke(writing_prompt) return {"final_output": response.content}现在我们把这些节点连接起来:
def create_writing_workflow(): # 创建状态图 workflow = StateGraph(WritingState) # 添加节点 workflow.add_node("analyze", analyze_request) workflow.add_node("outline", generate_outline) workflow.add_node("write", write_content) # 定义流程 workflow.add_edge(START, "analyze") workflow.add_edge("analyze", "outline") workflow.add_edge("outline", "write") workflow.add_edge("write", END) # 编译图 app = workflow.compile() return app5. 测试运行
def main(): # 创建工作流 app = create_writing_workflow() # 准备初始状态 initial_state = { "messages": [{"role": "user", "content": "帮我写一篇关于Python学习心得的博客文章"}], "content_type": "", "tone": "", "length": "", "draft": "", "final_output": "" } # 运行工作流 result = app.invoke(initial_state) print("=== 最终输出 ===") print(result["final_output"])if __name__ == "__main__": main()def should_revise(state: WritingState): """判断是否需要修订内容""" # 这里可以添加更复杂的判断逻辑 # 比如内容质量检查、长度检查等 content_length = len(state["final_output"]) if content_length < 200: return "revise" else: return "end"def revise_content(state: WritingState): """修订内容""" revise_prompt = f""" 原始内容:{state['final_output']} 这个内容似乎太短了,请扩展并改进内容, 使其更加详细和有用。 """ response = llm.invoke(revise_prompt) return {"final_output": response.content}# 在工作流中添加条件分支def create_advanced_workflow(): workflow = StateGraph(WritingState) # 添加所有节点 workflow.add_node("analyze", analyze_request) workflow.add_node("outline", generate_outline) workflow.add_node("write", write_content) workflow.add_node("revise", revise_content) # 定义流程 workflow.add_edge(START, "analyze") workflow.add_edge("analyze", "outline") workflow.add_edge("outline", "write") # 添加条件分支 workflow.add_conditional_edges( "write", should_revise, { "revise": "revise", "end": END } ) workflow.add_edge("revise", END) return workflow.compile()状态是整个流程的核心,设计时要考虑所有需要传递的信息
使用TypedDict可以获得更好的代码提示和类型检查
2. 错误处理
def safe_node_execution(func): """装饰器:为节点添加错误处理""" def wrapper(state): try: return func(state) except Exception as e: print(f"节点执行出错:{e}") return {"error": str(e)} return wrapper@safe_node_executiondef your_node_function(state): # 你的节点逻辑 pass在开发过程中,你可以添加日志来跟踪流程:
def debug_state(state: WritingState): """调试节点:打印当前状态""" print("=== 当前状态 ===") for key, value in state.items(): if isinstance(value, str) and len(value) > 100: print(f"{key}: {value[:100]}...") else: print(f"{key}: {value}") print("==================") return {}复杂的对话系统:需要记住上下文,根据不同情况选择不同回复策略
数据处理流水线:多步骤的数据清洗、分析、生成报告
内容创作工具:像我们刚才的例子,需要分析、规划、创作、修订的多步流程
客户服务自动化:根据问题类型路由到不同处理流程
建议大家从简单的线性流程开始,逐步添加条件分支和复杂逻辑。记住,好的工作流设计关键在于:
清晰的状态定义
职责单一的节点函数
合理的流程控制
53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费POC验证,效果达标后再合作。零风险落地应用大模型,已交付160+中大型企业
2026-01-29
自建一个 Agent 很难吗?一语道破,万语难明
2026-01-28
全球首个Skills Vibe Agents,AtomStorm技术揭秘:我是怎么用Context Engineering让Agent不"变傻"的
2026-01-22
Deepagents落地场景来了:用openwork实现专属办公小管家
2026-01-05
快速上手:LangChain + AgentRun 浏览器沙箱极简集成指南
2026-01-05
为什么大模型企业都在强调可以连续工作XX小时的Agent和模型?长时运行Agent解析(Long-Running Agents)
2025-12-29
单agent落幕,双agent才能解决复杂问题!附LangGraph+Milvus实操
2025-12-26
跟我学LangChain:提示词模板,PromptTemplate包装器,工程化管理你的提示词
2025-12-24
别再堆 Prompt 了:用 LangChain 1.0 搭建“深度思考 Agent”
2025-11-06
2025-12-21
2025-12-21
2025-11-25
2025-12-08
2025-11-08
2025-11-18
2025-11-07
2025-11-25
2025-11-19
2025-11-03
2025-10-29
2025-07-14
2025-07-13
2025-07-05
2025-06-26
2025-06-13
2025-05-21