尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

LangChain V1.3 + LangGraph 多智能体开发实战:架构设计与部署

LangChain V1.3 + LangGraph 多智能体开发实战:架构设计与部署 LangChain V1.3 和 LangGraph 的组合基本是目前企业级 AI Agent 开发绕不开的技术栈。如果说 LangChain 解决的是“怎么把模型接进来、怎么组织提示词和工具调用”那 LangGraph 解决的就是“多个智能体怎么协作、状态怎么流转、任务怎么编排”。这篇文章不会讲概念堆砌而是直接拆开 LangChain V1.3 LangGraph 的核心组件、多智能体架构设计和可运行的代码实战帮你快速建立一套能落地到企业项目里的 AI Agent 开发框架。整个技术栈最值得关注的能力有四个一是 LangGraph 的状态图机制把原来难以控制的 Agent 多轮调用变成了可视化、可断点、可恢复的流程编排二是多智能体协作不同角色智能体可以共享状态、按条件切换执行路径三是 LangChain V1.3 对工具调用和模型接入的统一抽象换了底层模型不影响上层业务逻辑四是 API 服务化LangGraph 可以暴露 REST 接口批量任务也能直接驱动。下面会带大家从环境准备、安装部署、核心组件分析到多智能体代码实战、接口调用和常见问题排查完整过一遍。1. LangChain V1.3 LangGraph 核心能力速览能力项说明项目类型开源 AI Agent 应用开发框架Python 为主核心价值模型接入统一抽象 多智能体状态图编排V1.3 重点和 LangGraph 深度集成工具调用、结构化输出、可观测性增强主要功能Agent 构建、工具调用、多智能体协作、RAG、工作流编排、状态持久化、断点恢复硬件门槛框架本身对硬件要求低远程模型 API 无需 GPU本地模型部署需按模型规格准备显卡显存支持平台Windows / Linux / macOS 均可建议 Linux 服务器用于生产启动方式Python 脚本启动、LangGraph 开发服务器、Docker 部署是否支持 API支持LangGraph Server 可暴露 REST API是否支持批量任务支持通过循环调度或工作流状态机批量处理任务适合场景企业知识库问答、自动化运维助手、客服工单处理、多步骤业务流自动化2. 多智能体系统适用场景与使用边界多智能体架构最适合解决这类问题一个任务需要拆成多个专业角色协作完成并且每个角色有独立的上下文和工具权限。典型场景包括智能客服系统意图识别 Agent 先判断用户问题再分流给售后 Agent、技术 Agent 或退换货 Agent各自调用对应业务接口。自动化竞品分析数据采集 Agent 抓取网页内容分析 Agent 提炼要点报告生成 Agent 输出结构化文档。企业知识库问答检索 Agent 负责从向量库召回推理 Agent 负责整合答案审核 Agent 负责校验结果是否严谨。工单自动处理分类 Agent 判断工单类型处理 Agent 执行操作通知 Agent 生成回复并发送。使用边界要特别注意。LangGraph 本身是通用编排框架不等于业务逻辑本身它接管的是“流程控制权”但最终的操作对象可能是数据库、文件系统、第三方 API。这意味着权限设计必须从架构层面控制每个 Agent 只能调用最小权限的工具涉及人脸、声音、用户隐私数据、版权素材时必须确认授权和合规边界。多智能体系统还容易出现“级联幻觉”也就是上游 Agent 的错误结果被下游 Agent 当成事实继续加工所以关键节点建议增加人工审核或自动校验步骤。3. LangChain V1.3 与 LangGraph 本地部署环境准备3.1 操作系统与基础环境LangChain 和 LangGraph 都是 Python 库对操作系统没有特殊限制。本地开发建议 Windows 10/11 或 macOS生产环境建议 Linux 服务器。需要提前准备Python 3.9 到 3.12 版本。LangGraph 对 Python 版本有要求过旧的 3.8 或最新的 3.13 都可能出现依赖兼容问题。pip 包管理工具。Git用于克隆示例项目。一个可用的模型 API Key比如 OpenAI、通义千问、智谱、DeepSeek、Ollama 本地模型等。3.2 模型接入方式LangChain V1.3 的核心优势之一是模型接入抽象。同一个业务代码可以切换不同模型提供商。常见方式远程 API 调用需要配置 API Key适合生产环境无 GPU 要求。本地模型调用使用 Ollama 或 vLLM 部署开源模型需要准备显卡。显存占用取决于模型规模7B 级量化模型一般需要 6GB 以上显存13B 级模型建议 12GB 以上。LangChain 框架本身不直接占用显存显存资源来自底层模型服务。更稳妥的判断是如果只做框架学习和流程开发Laptop 上加远程 API Key 就够了不需要关注显存只有本地模型部署才需要评估显卡。3.3 磁盘与端口LangGraph 开发服务器和依赖包占用空间不大预留 2GB 空间即可。如果是本地模型部署则按模型文件大小准备额外空间。默认端口为 2024 和 8123如果本地端口被占用需要在启动时指定其他端口。4. LangChain V1.3 LangGraph 安装部署与启动方式4.1 安装依赖建议先创建虚拟环境避免和系统 Python 环境冲突。python -m venv langgraph-env source langgraph-env/bin/activate # Windows 使用 langgraph-env\Scripts\activate pip install --upgrade pip pip install langchain langgraph langchain-openai如果计划使用 LangGraph 的开发调试服务器和可视化界面还需要安装pip install langgraph-cli[inmem] langchain-cli需要说明的是版本号会持续更新。安装时可以直接用最新稳定版V1.3 是当前周期内的关键版本重点强化了和 LangGraph 的集成能力。4.2 验证安装python -c import langchain; print(langchain.__version__) python -c import langgraph; print(langgraph.__version__)如果能正常输出版本号说明安装成功。4.3 LangGraph 开发服务器启动LangGraph 提供了本地开发服务器可以可视化查看状态图也可以直接调试 Agent。项目根目录创建langgraph.json配置文件{ dependencies: [.], graphs: { agent: ./agent.py:graph }, env: .env }然后在项目目录启动langgraph dev启动后访问http://127.0.0.1:2024可以看到可视化界面。注意这里的启动方式和uvicorn app直接启动的区别langgraph dev是 LangGraph 自带的开发模式会加载状态图配置并开启图形化调试界面uvicorn app是启动普通的 Python Web 服务适用于已经封装好的 API 服务。两者用途不同开发调试阶段推荐先用langgraph dev。5. 核心组件剖析从 Chain 到 Graph5.1 LangChain V1.3 核心组件吃透 LangChain V1.3不需要把每个 API 背下来关键是理解组件之间的协作关系。ChatModel大模型聊天接口的统一封装。不管底层是 OpenAI、通义千问还是本地 Ollama上层调用方式一致。Prompt Templates提示词模板支持变量插值和结构化输出约束。Tools工具抽象任何 Python 函数只要加上装饰器就能变成 Agent 可调用的工具。工具的定义决定了 Agent 的边界。Memory对话记忆管理实现多轮上下文保留。OutputParser输出解析器把模型返回的文本转成 JSON、字典等结构化数据或者直接绑定模型的工具调用能力。5.2 LangGraph 核心组件LangGraph 的核心是状态图StateGraph。它把一次 Agent 运行建模成一个图节点是处理步骤边是步骤间的流转关系。核心概念包括State全局状态对象所有节点共享。每次节点执行后可以更新状态。Node处理单元可以是大模型调用、工具执行、条件判断或任何自定义 Python 函数。Edge节点之间的连接线分为普通边和条件边。Conditional Edge条件边根据当前状态决定下一步走向这是多智能体分流的关键。START / END图的起点和终点。Checkpointer状态持久化和断点机制。支持把运行状态保存下来中断后从断点恢复。LangGraph Server把图编译成可部署的 API 服务支持多客户端并发访问。理解 LangGraph核心是理解状态流转每条消息、每个工具结果都写入 State下一个节点从 State 中读取输入执行完又更新 State。这种设计让复杂的多智能体协作变得可控。5.3 LangChain 和 LangGraph 的区别很多初学者会混淆两个项目。简单说LangChain 解决的是“模型、工具、记忆、提示词”这些组件的标准化问题它提供的是积木LangGraph 解决的是“这些积木怎么组装成一个状态化、可恢复的工作流”的问题它提供的是流水线和工位。在 AI Agent 开发中两者通常配合使用LangChain 负责模型和工具的接入层LangGraph 负责流程编排层。6. 多智能体架构设计与代码实战6.1 确定多智能体协作模式多智能体常见协作模式有三种主管-工人模式主管 Agent 负责拆解任务分发给多个工人 Agent 执行再汇总结果。适合任务可并行拆分的场景。流水线模式Agent 按顺序执行上一个 Agent 的输出是下一个 Agent 的输入。适合有明确先后依赖的任务。自主协商模式多个 Agent 自由交流互相传递消息直到达成一致。适合复杂谈判或生成任务但可控性较弱。企业级开发更推荐前两种代码易于维护状态可监控出现错误也好定位。6.2 基础 Agent 实战一个带工具的客服 Agent先写一个最简可运行的 Agent。这个 Agent 有两个能力查订单状态和计算退款金额。from typing import TypedDict from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list final_answer: str tool def query_order_status(order_id: str) - str: 查询订单状态输入订单号返回订单当前状态。 # 这里仅为示例实际接入业务系统需替换 status_map {A1001: 已发货, A1002: 待付款, A1003: 已完成} return status_map.get(order_id, 未找到该订单) tool def calculate_refund(order_id: str) - str: 计算退款金额输入订单号返回可退金额。 amount_map {A1001: 299.0, A1002: 0.0, A1003: 1299.0} return f订单 {order_id} 可退款金额为 {amount_map.get(order_id, 0.0)} 元 model ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyyour-api-key, ) tools [query_order_status, calculate_refund] model_with_tools model.bind_tools(tools) def call_model(state: AgentState) - AgentState: response model_with_tools.invoke(state[messages]) return {messages: [response]} def call_tool(state: AgentState) - AgentState: last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: return state responses [] for tc in tool_calls: tool_name tc[name] tool_args tc[args] if tool_name query_order_status: result query_order_status.invoke(tool_args) elif tool_name calculate_refund: result calculate_refund.invoke(tool_args) else: result 未知工具 responses.append({role: tool, content: result, tool_call_id: tc[id]}) return {messages: responses} def should_continue(state: AgentState) - str: last_message state[messages][-1] if hasattr(last_message, tool_calls) and len(last_message.tool_calls) 0: return action return end graph_builder StateGraph(AgentState) graph_builder.add_node(agent, call_model) graph_builder.add_node(action, call_tool) graph_builder.add_edge(START, agent) graph_builder.add_conditional_edges( agent, should_continue, {action: action, end: END} ) graph_builder.add_edge(action, agent) graph graph_builder.compile()运行测试result graph.invoke( {messages: [{role: user, content: 帮我查一下订单 A1001 的状态顺便算一下能退多少钱}]} ) print(result[messages][-1].content)这个例子中bind_tools让模型具备了工具调用能力should_continue作为条件边控制是继续调用工具还是结束。这是 LangGraph 里最基础也最核心的 Agent 循环。6.3 多智能体实战主管-工人模式接下来实现一个主管-工人模式。主管 Agent 负责理解用户意图决定是调用“数据分析 Agent”还是“客服知识库 Agent”。from typing import TypedDict, Literal from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END class MultiAgentState(TypedDict): input_text: str next_agent: str data_result: str support_result: str final_output: str data_agent_prompt 你是一个数据分析助手擅长处理数据统计和报表生成。 用户的问题是{input_text} 请输出分析结果。 support_agent_prompt 你是一个客服知识库助手擅长解答售后、退款、物流类问题。 用户的问题是{input_text} 请输出标准解答。 manager_model ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyyour-api-key, ) def manager_agent(state: MultiAgentState) - MultiAgentState: classification_prompt f 请判断以下问题应该分发给哪个助手处理。 如果问题涉及数据统计、报表、趋势分析输出 data。 如果问题涉及客服售后、退款、物流输出 support。 只能输出 data 或 support不要输出其他内容。 用户问题{state[input_text]} response manager_model.invoke(classification_prompt) agent_name response.content.strip().lower() return {next_agent: agent_name} def data_agent(state: MultiAgentState) - MultiAgentState: response ChatOpenAI( modelgpt-4o-mini, api_keyyour-api-key, temperature0, ).invoke(data_agent_prompt.format(input_textstate[input_text])) return {data_result: response.content} def support_agent(state: MultiAgentState) - MultiAgentState: response ChatOpenAI( modelgpt-4o-mini, api_keyyour-api-key, temperature0, ).invoke(support_agent_prompt.format(input_textstate[input_text])) return {support_result: response.content} def route_agent(state: MultiAgentState) - Literal[data_agent, support_agent]: return state[next_agent] if state[next_agent] in (data_agent, support_agent) else support_agent def final_node(state: MultiAgentState) - MultiAgentState: if state.get(data_result): return {final_output: state[data_result]} return {final_output: state[support_result]} builder StateGraph(MultiAgentState) builder.add_node(manager, manager_agent) builder.add_node(data_agent, data_agent) builder.add_node(support_agent, support_agent) builder.add_node(final, final_node) builder.add_edge(START, manager) builder.add_conditional_edges(manager, route_agent, {data_agent: data_agent, support_agent: support_agent}) builder.add_edge(data_agent, final) builder.add_edge(support_agent, final) builder.add_edge(final, END) multi_graph builder.compile()测试时分别输入一个数据类问题和一个客服类问题观察路由是否正确切换。print(multi_graph.invoke({input_text: 帮我统计上个月销售数据趋势})) print(multi_graph.invoke({input_text: 我的订单发货了没有}))这就是多智能体架构的核心状态图决定流程条件边决定分流每个 Agent 只负责自己的专业领域整体状态在节点间传递。7. 接口 API 调用与批量任务7.1 编译成 API 服务在企业项目中通常不会直接在业务代码里调用graph.invoke()而是把图发布成 API 服务。LangGraph 提供 LangGraph Server编译过的图可以快速变成 REST 接口。创建agent.py并在文件底部导入编译后的图对象graph graph_builder.compile()然后在langgraph.json中配置入口{ dependencies: [.], graphs: { customer_agent: ./agent.py:graph } }启动服务langgraph serve --port 81237.2 Python 调用示例API 服务启动后可以用 Python 脚本调用接口。import requests url http://127.0.0.1:8123/customer_agent/invoke payload { input: { messages: [ {role: user, content: 帮我查一下订单 A1001 的状态} ] }, config: { configurable: { thread_id: test-001 } } } response requests.post(url, jsonpayload, timeout60) print(response.json())需要说明的是不同版本的 LangGraph Server 接口路径可能有差异实际使用时以langgraph serve启动后输出的路由文档为准。如果接口返回 404优先检查路由配置文件中的图名称是否和启动时一致。7.3 批量任务设计批量任务可以通过循环调用接口实现但更推荐在工作流内部做批处理。给一个通用模板import time import json import requests api_url http://127.0.0.1:8123/customer_agent/invoke tasks [ {id: task1, question: 退款多久到账}, {id: task2, question: 物流信息不更新怎么办}, {id: task3, question: 可以开发票吗}, ] results {} for task in tasks: payload { input: { messages: [{role: user, content: task[question]}] }, config: {configurable: {thread_id: task[id]}} } try: resp requests.post(api_url, jsonpayload, timeout60) results[task[id]] { status: success, answer: resp.json() } except Exception as e: results[task[id]] { status: failed, error: str(e) } # 控制请求频率不要连续打爆服务 time.sleep(0.5) with open(batch_results.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(json.dumps(results, ensure_asciiFalse, indent2))批量任务的关键是加日志、加失败重试、控制并发。更稳妥的处理是先用少量样本跑通全链路再扩大批量规模。8. 资源占用与性能观察LangChain V1.3 LangGraph 框架本身对内存和 CPU 的占用很低核心资源消耗集中在底层模型服务上。可以从几个维度观察远程 API 模式本机只跑 Python 进程内存占用一般在几百 MB 以内无显存压力。本地模型模式使用 Ollama 或 vLLM 部署模型时显存占用由模型大小和并发数决定。观察方式是在服务端执行nvidia-smi查看 GPU 利用率。多智能体并发每个并发任务都会创建独立的图执行实例状态数据会暂存在内存中。线程数、状态大小、工具执行时长都会影响整体吞吐量。长对话场景State 里累积的历史消息越多每次模型调用消耗的 token 越多响应变慢。建议定期裁剪历史消息。框架层面的性能优化手段包括使用流式输出避免等待完整响应。对工具执行结果做缓存避免重复调用外部 API。在批量任务中限制并发数避免触发模型服务的速率限制。使用 Checkpointer 做状态持久化而不是把所有历史都放在内存里。9. 常见问题与排查方法问题现象可能原因排查方式解决方案安装时依赖冲突Python 版本不兼容或包版本锁冲突查看 pip 报错信息升级 Python 到 3.10-3.12重装依赖导入 langchain 失败安装不完整或虚拟环境未激活pip list查看已安装包重新执行安装命令激活正确的虚拟环境模型调用报 401API Key 错误或未配置环境变量检查.env文件和 Key 有效期更新 Key确认环境变量是否被正确加载Agent 不调用工具模型版本不支持 tool calling检查模型文档和bind_tools写法更换支持工具调用的模型或改用提示词强制输出图运行卡住不结束条件边返回值未匹配到对应路由检查条件函数返回值和边配置字典打印条件函数返回值确保 key 一致LangGraph 服务启动后接口 404langgraph.json 图名称和请求路径不一致查看启动日志中的路由表按实际路由调整 URL 或配置文件批量任务部分失败外部 API 限流或超时查看错误日志中的状态码添加重试机制、降低并发、加大超时时间显存不足本地模型规模过大或并发过高nvidia-smi查看显存占用换小尺寸模型、降低并发、启用量化端口被占用执行了多个未正常关闭的服务检查进程列表杀掉残留进程或更换启动端口状态数据丢失未配置 Checkpointer 或线程 ID 不统一检查每次调用的 config 参数配置持久化存储统一thread_id10. 最佳实践与使用建议做 LangChain V1.3 LangGraph 企业级开发建议从第一天就按工程标准来组织代码。第一先小参数跑通全链路再扩展。第一次不要直接上复杂多智能体先写一个单节点图输入输出正确后再加条件边、加工具、加子 Agent。第二保留一套最小可运行配置。下载或创建示例项目后第一时间提交到自己的 Git 仓库记录可运行的依赖版本清单避免后续升级导致环境不可复现。第三模型、输入、输出分目录管理。项目目录建议这样组织project/ ├── agent.py # 图定义和节点逻辑 ├── tools/ # 工具函数 ├── graphs/ # 不同业务场景的图 ├── tests/ # 测试用例 ├── data/inputs/ # 批量任务输入 ├── data/outputs/ # 批量任务输出 ├── logs/ # 运行日志 ├── .env # 环境变量不进 Git └── langgraph.json # LangGraph 服务配置第四批量任务必须加日志和失败重试。每个任务记录开始时间、结束时间、状态码、错误信息失败任务进入重试队列重试仍失败再进入人工处理队列。第五接口服务必须限制访问范围。生产环境不要直接把 LangGraph Server 暴露到公网建议放在内网通过 API 网关做鉴权按用户或按调用方设置速率限制。第六涉及数据隐私和版权素材时必须在系统设计层面控制权限。多智能体系统会比单 Agent 触及更多数据源建议为不同 Agent 配置独立的工具权限和数据集访问范围并对敏感操作增加人工审批节点。声音、人脸、版权文档的使用要获得对应授权发布或商用前要做效果复核。第七关注 LangGraph 状态持久化的线程 ID 设计。在线客服、多轮对话等场景中thread_id相当于会话 ID同一会话的所有消息必须使用同一个 ID否则上下文会丢失。第八善用 LangGraph 的断点机制做人工审核。在关键节点设置 breakpoint图运行到该处会暂停人工确认后才继续执行这对涉及资金操作、对外发送消息的场景尤其重要。11. 总结与下一步这一套 LangChain V1.3 LangGraph 的组合拳最值得先尝试的就是基础的带工具 Agent让模型学会调用一个查询函数然后观察 LangGraph 的条件边是怎么把“模型想调工具”变成“真去执行工具”的。跑通这一步再去扩展主管-工人模式的多智能体系统难度会小很多。最容易踩的坑是版本兼容和路由配置。优先用 Python 3.10 到 3.12 建虚拟环境写完图之后先本地跑一次graph.invoke()确认状态流转正常再启动langgraph dev看可视化界面最后才考虑发布成 API 服务。整体流程从本地测试到接口暴露每一步都有迹可循不会出现“看起来配好了但一调就挂”的情况。接下来可以继续验证的方向包括把 LangGraph 和 RAG 检索链路整合成一个完整的知识库问答 Agent接入企业内部的业务工具让 Agent 真正具备执行能力在 LangGraph 中引入人工审核节点跑通“自动化执行、关键步骤人工确认”的混合流程尝试用 Checkpointer 实现长时间运行的多轮任务恢复。这些方向上LangChain V1.3 LangGraph 都能提供比较完整的基础设施支撑代码模型一旦跑通后续扩展就是加节点和加边的事。
返回列表