微信扫码
添加专属顾问
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-07-01
LangGraph Runtime 是什么?一文讲清Runtime与Context的作用与用法!
2026-06-26
拆解Agent Harness的11大核心组件与工程实践(附下载)
2026-06-05
让 Agent 快速上生产:基于 OceanBase 和 LangChain 打造的智能体系统解决方案发布
2026-05-19
90% 的 Agent 失败,不是框架不行,而是卡在 5 个工程问题
2026-05-14
用两行代码将 AgentRun 集成到你的应用
2026-05-06
LangChain 深度智能体(Deep Agents)入门
2026-04-19
万字讲透Agent Harness的十二大模块
2026-04-08
同一个模型,换个Harness排名跳了25位:智能体基础设施完全解剖
2026-04-19
2026-04-08
2026-05-06
2026-05-19
2026-05-14
2026-06-05
2026-06-26
2026-07-01
2026-03-26
2025-11-03
2025-10-29
2025-07-14
2025-07-13
2025-07-05
2025-06-26
2025-06-13
欢迎您使用【53AI 官方网站】(以下简称“本网站”或“我们”)。本《会员服务协议》(以下简称“本协议”)是您(以下简称“会员”或“用户”)与【深圳市博思协创网络科技有限公司】之间关于注册、登录及使用本网站会员服务所订立的法律协议。
在您注册或登录前,请务必审慎阅读、充分理解各条款内容,特别是免除或限制责任的条款、知识产权条款、争议解决条款等。此类条款将以加粗形式提示您注意。 当您通过微信公众号授权、手机验证码验证或其他方式成功登录本网站时,即视为您已完全理解并同意接受本协议的全部内容。
一、 定义
本网站:指由【深圳市博思协创网络科技有限公司】运营的,域名为【53ai.com】的网站及相关移动端页面。
会员服务:指本网站向注册会员提供的知识库文章查阅、内容检索及其他相关增值服务。
知识库内容:指本网站发布的包括但不限于文字、图表、数据、研究报告、行业分析等数字化内容资源。
二、 账号注册与登录
登录方式:本网站支持以下登录方式,您可根据实际情况选择:
微信公众号授权登录:您同意将您的微信OpenID信息授权给本网站,用于创建或关联会员账号。
手机验证码登录:您需提供真实有效的手机号码,并通过短信验证码完成身份验证与登录/注册。
账号安全:您的账号仅限您本人使用,禁止赠与、借用、租用、转让或售卖。因您保管不善导致的账号被盗、密码泄露等损失,由您自行承担。
实名认证:根据相关法律法规要求,我们可能要求您在特定功能下完成实名认证。如您拒绝提供,可能无法使用部分或全部服务。
未成年人保护:若您未满18周岁,请在法定监护人的陪同下阅读本协议,并在征得监护人同意后使用本服务。
三、 服务内容与规范
知识库查阅权限:会员登录后,有权按照其会员等级对应的权限范围,在线浏览、检索本网站知识库中的相关文章及内容。
服务变更:我们有权根据业务发展需要,调整、变更或终止部分服务内容,并将以网站公告、公众号消息等方式提前通知。
禁止行为:您在使用服务时不得实施以下行为:
利用技术手段批量爬取、下载、转存知识库内容;
将知识库内容用于商业目的或未经授权地向第三方传播;
干扰本网站正常运行或侵犯其他用户合法权益;
发布违法违规信息或从事违反公序良俗的活动。
四、 知识产权声明
权利归属:本网站知识库中的排版设计、软件代码等内容的知识产权均归【公司全称】或原权利人所有,受《中华人民共和国著作权法》等法律保护。
有限许可:本网站授予会员一项非独占、不可转让、不可转授权的普通许可,仅限于个人学习、研究之目的在线查阅知识库内容。
侵权追责:未经书面许可,任何单位或个人不得以任何形式复制、转载、摘编、镜像、汇编或以其他方式使用上述内容。一经发现,我们保留追究其法律责任的权利。
五、 个人信息保护
我们重视对您个人信息的保护。关于我们如何收集、使用、存储和保护您的个人信息,请单独阅读 《隐私政策》。
您通过微信公众号授权或手机号验证所提供的信息,我们将严格按照《个人信息保护法》的规定处理,仅用于身份识别、服务提供及安全验证等必要用途。
您可以随时通过网站设置或联系客服行使查阅、更正、删除个人信息及撤回授权同意的权利。
六、 免责声明
内容准确性:知识库内容仅供参考,不构成专业建议。我们不对其完整性、准确性、时效性作任何明示或暗示的保证,您应自行判断并承担使用风险。
不可抗力:因自然灾害、政策法规变化、网络故障、第三方平台接口异常(如微信接口维护、运营商短信通道故障)等不可抗力导致的服务中断或延迟,我们不承担违约责任。
第三方链接:本网站可能包含指向第三方网站的链接,该等网站的内容和服务不受我们控制,请您自行甄别风险。
七、 违约责任
如您违反本协议约定,我们有权视情节采取警告、限制功能、暂停服务、注销账号等措施,并保留要求赔偿损失的权利。
如因您的违约行为导致我们遭受行政处罚、第三方索赔或商誉损失,您应承担全部赔偿责任(包括但不限于罚款、赔偿金、律师费、公证费等)。
八、 法律适用与争议解决
本协议的订立、执行和解释均适用中华人民共和国大陆地区法律。
因本协议产生的或与本协议有关的任何争议,双方应友好协商解决;协商不成的,任何一方均可向【公司所在地】有管辖权的人民法院提起诉讼。
九、 其他
本协议构成双方就本服务达成的完整协议,取代此前任何口头或书面约定。
本协议任一条款被认定为无效或不可执行的,不影响其他条款的效力。
我们对本协议享有最终解释权,并在法律允许的范围内保留随时修改的权利。修改后的协议一经公布即生效,继续使用服务即视为同意修订内容。