尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

LangChain与LangGraph实战:从零构建AI智能体工作流

LangChain与LangGraph实战:从零构建AI智能体工作流 1. 项目概述为什么现在是“Agent时代”如果你最近在AI开发社区里泡着应该能明显感觉到一股热潮大家讨论的焦点正从单纯地调用大模型API转向构建能够自主规划、使用工具、并持续运行的“智能体”。这不仅仅是概念炒作而是技术栈成熟度达到拐点后的必然趋势。过去我们想让一个AI程序去网上查天气、然后根据天气建议我穿什么衣服可能需要写一堆复杂的if-else逻辑手动拼接提示词还得处理各种API调用的异常。现在有了像LangChain和LangGraph这样的框架构建这样的“智能体”变得前所未有的模块化和直观。简单来说LangChain提供了一个丰富的“工具箱”和一套组装逻辑让你能轻松地将大模型、知识库、计算工具、搜索引擎等连接起来。而LangGraph则是在此基础上引入了“图”的概念专门用来构建那些有状态、能循环、可分支的复杂智能体工作流。你可以把LangChain看作是给你提供了乐高积木块和基本的拼接说明书而LangGraph则允许你设计并搭建一个带有齿轮、传送带和反馈回路的自动化工厂。这个项目标题“开启你的Agent时代”精准地捕捉了当下的机遇。对于开发者而言无论你是想做一个能自动处理客服工单的助手一个能调研市场并生成报告的分析师还是一个能24小时监控系统日志并自主排障的运维专家掌握LangChain和LangGraph都意味着你拿到了进入下一代AI应用开发大门的钥匙。而“语言堆栈抉择”则点出了一个非常现实的问题作为开发者你是用Python还是TypeScript来开启这段旅程这不仅仅是个人偏好更关乎项目类型、团队协作和生态系统的选择。接下来我会结合我自己的踩坑经验带你从零开始拆解这个开发入门过程并帮你做出那个关键的技术选型决定。2. 核心架构解析LangChain与LangGraph的角色与协同在深入代码之前我们必须先厘清这两个核心框架的定位和它们是如何协同工作的。很多新手容易混淆觉得学了LangChain是不是就不用学LangGraph了或者反过来。其实它们是互补关系分别解决了智能体开发中不同层面的问题。2.1 LangChain智能体的“工具箱”与“粘合剂”LangChain的核心价值在于“链”。它将大模型与其他数据源或功能连接起来形成一个可执行的序列。你可以把它理解为一个高度抽象的工作流编排器但其最初的设计更侧重于线性的、确定性的链条。核心组件包括Models:支持各种LLM如OpenAI GPT、Anthropic Claude、本地部署模型和Embedding模型。Prompts:提供模板管理、少量示例Few-shot提示等功能让提示工程更规范。Indexes:用于连接外部数据最常见的就是RAG检索增强生成中的文档加载、切分、向量化存储与检索。Memory:为对话或链提供短期或长期的记忆能力比如保存历史对话。Chains:这是核心将上述组件按顺序组合起来。例如一个简单的链可能是“接收用户问题 - 检索相关文档 - 将问题和文档组合成提示词 - 发送给LLM - 返回答案”。Agents:LangChain的Agent是一个高级抽象它让LLM能够动态地决定为了完成一个任务需要调用哪些工具Tools并按什么顺序调用。这是实现“自主性”的关键。然而传统的LangChain Agent虽然强大但在描述复杂的工作流时尤其是那些包含循环、条件分支、并行执行或需要精细状态管理的工作流时会显得有些力不从心。它的执行路径更多是由LLM在每一步实时“思考”决定的虽然灵活但可控性和可观测性对于复杂场景来说是个挑战。2.2 LangGraph智能体的“状态机”与“流程图”LangGraph应运而生它建立在LangChain之上引入了有向图的计算模型。在这个模型中节点Node代表一个可执行的操作比如调用一次LLM或运行一个工具边Edge代表操作之间的流转条件。它的核心优势在于显式的工作流定义你需要像画流程图一样明确定义整个智能体的所有步骤和跳转逻辑。这带来了极佳的可读性和可维护性。新成员看一眼图就能理解智能体的整体逻辑。强大的状态管理LangGraph有一个核心的State概念它是一个共享的字典在所有节点之间传递和更新。你可以清晰地定义状态的结构每个节点读取什么、修改什么一目了然。这对于需要多步骤信息聚合的场景至关重要。支持复杂控制流循环、条件分支、并行、异步执行这些在LangGraph的图模型中可以很自然地表达。例如你可以设置一个“审核”节点如果LLM生成的内容不达标就沿着一条边跳回“重写”节点形成循环。持久化与人类介入LangGraph原生支持将工作流状态持久化这意味着你可以暂停一个长任务稍后恢复。更重要的是它设计了“中断”机制可以让工作流在特定节点暂停等待人类输入或审核然后再继续这为安全可控的AI应用提供了可能。所以两者的关系可以概括为你用LangChain提供的丰富组件Models, Tools, Prompts等来“武装”你的智能体然后用LangGraph的图模型来“设计”和“驱动”这个智能体的复杂行为逻辑。在LangGraph的节点里你大量使用的依然是LangChain的链、工具和代理。注意不要认为LangGraph是来替代LangChain Agent的。对于简单、线性的工具调用任务传统的LangChain Agent可能更快捷。但对于企业级、需要稳定可靠和复杂逻辑的智能体应用LangGraph是更专业的选择。它们是一个生态体系内的不同工具。3. 开发环境搭建与语言堆栈深度抉择这是项目启动的第一步也是决定后续开发体验的关键一步。Python和TypeScript运行在Node.js环境是LangChain/LangGraph官方支持最好的两大语言生态。我们分别来看。3.1 Python 生态数据科学与AI研究的“主场”优势生态绝对统治力绝大多数机器学习、深度学习的库PyTorch, TensorFlow、数据科学工具pandas, numpy以及新兴的AI模型和工具都率先或仅提供Python接口。如果你想对智能体进行深度定制例如微调嵌入模型、集成特定的科研模型Python是唯一选择。Jupyter Notebook友好非常适合快速原型验证、实验和数据分析。你可以交互式地测试链的每一个环节直观看到输出。社区与教程资源最丰富遇到问题时在Stack Overflow、GitHub或各类博客上找到Python相关的解决方案和案例的概率最大。LangChain功能最全通常LangChain的新特性会先在Python版本上推出和完善。劣势与避坑指南“依赖地狱”与虚拟环境Python的包管理是个老生常谈的问题。不同项目对同一包的不同版本依赖可能导致冲突。必须使用虚拟环境强烈推荐使用conda或venv。对于AI项目conda在管理一些带有C扩展的复杂依赖如某些CUDA版本的PyTorch时更有优势。# 使用 conda conda create -n my-langgraph-agent python3.11 conda activate my-langgraph-agent # 使用 venv python -m venv venv # 在Windows上: venv\Scripts\activate # 在Mac/Linux上: source venv/bin/activate部署复杂度将Python应用打包成可独立部署的服务相比Node.js稍显复杂。虽然Docker化是标准解决方案但镜像体积通常较大。典型安装命令pip install langchain langchain-community langgraph # 如果你要使用OpenAI的模型还需要 pip install openai # 可能用到的其他常用包 pip install pydantic chromadb tiktoken3.2 TypeScript/Node.js 生态全栈与云原生应用的“利器”优势前后端同构如果你的团队主要技术栈是JavaScript/TypeScript或者你正在构建一个需要与前端React, Vue等紧密交互的AI应用比如一个实时聊天的AI助手界面使用TS可以共享类型定义、减少上下文切换实现真正的全栈开发。部署与运维轻量Node.js应用启动快、内存占用相对较小非常适合构建轻量级API服务或Serverless函数如Vercel, AWS Lambda。部署流程在现代前端工具链下非常顺畅。工具链与工程化TypeScript的静态类型检查能在开发阶段捕获大量潜在错误配合ESLint、Prettier等工具代码质量和团队协作体验极佳。这对于中大型、多人协作的Agent项目至关重要。异步处理天然友好Node.js的非阻塞I/O模型非常适合Agent需要频繁调用外部API网络请求、数据库查询的场景。劣势与注意事项AI特定库的广度不足虽然核心的LangChain功能齐全但当你需要集成一个非常新的、小众的Python-only的AI库时可能会找不到对应的TS版本或需要自己封装。原型验证速度可能稍慢虽然也有类似REPL的环境但相比Jupyter Notebook那种单元格即时代码执行和数据可视化的体验还是略有差距。典型安装命令# 初始化一个Node.js项目如果还没有package.json npm init -y # 安装核心依赖 npm install langchain/langgraph langchain/core # 安装社区工具和OpenAI集成 npm install langchain/community langchain/openai # 安装类型定义和开发工具 npm install --save-dev typescript types/node ts-node # 初始化tsconfig.json npx tsc --init3.3 如何抉择我的实战建议这不是一个非此即彼的选择但可以根据项目重心做决定选择 Python如果你的项目核心是研究、实验、模型微调或复杂的数据处理。你需要紧密集成Jupyter Notebook进行数据分析或教学演示。你的团队背景以数据科学家、算法工程师为主。你需要的某个关键工具或模型只有Python版本。选择 TypeScript如果你正在构建一个生产级的Web应用或API服务并且前端也是JS/TS技术栈。你的团队是全栈或后端工程师对JS生态更熟悉。你对代码类型安全、工程化规范和部署便捷性有很高要求。你的Agent逻辑相对稳定更侧重于工作流编排和业务集成而非底层模型魔改。高级策略混合架构在大型项目中一种越来越常见的模式是“TS前端/API层 Python AI核心层”。用Python实现最核心、最复杂的模型推理和AI逻辑并将其封装成gRPC或HTTP API服务。然后用TypeScript编写业务应用层调用这些服务。这结合了两者的优势但引入了服务间通信的复杂度。对于入门者我个人的建议是先从Python开始。因为其生态更成熟学习资料更多在探索和实验阶段遇到的阻力会更小。当你理解了Agent的核心概念后再根据项目需求评估是否切换到TS。事实上LangChain在两个语言中的核心概念高度一致切换成本并不高。4. 第一个LangGraph智能体实战构建一个“研究助手”原型理论说再多不如动手跑一遍。我们来构建一个简单的“研究助手”智能体。它的任务是给定一个研究主题它能自动联网搜索最新信息总结核心观点并生成一份简单的报告大纲。我们将使用Python环境进行演示。4.1 定义智能体的状态与节点首先我们需要定义智能体运行过程中需要共享的“状态”。这就像给智能体设计一个数据结构。from typing import TypedDict, List, Annotated import operator # 定义状态结构 class AgentState(TypedDict): # 用户输入的主题 topic: str # 搜索到的原始内容列表 raw_search_results: List[str] # 总结后的核心观点列表 summarized_points: List[str] # 最终生成的报告大纲 report_outline: str # 一个计数器用于演示循环控制 iteration_count: Annotated[int, operator.add] # 这个注解告诉LangGraph这个字段在多个节点中会被累加关键点Annotated[int, operator.add]是LangGraph的一个高级特性它声明iteration_count是一个可累加的整数。当多个节点修改它时LangGraph会自动执行加法操作而不是覆盖。这对于计数、汇总等场景非常有用。接下来我们定义图中的各个节点函数。每个节点接收一个state字典修改它并返回更新后的字典。from langchain_community.tools import TavilySearchResults from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate import os # 初始化工具和模型请确保已设置OPENAI_API_KEY和TAVILY_API_KEY环境变量 search_tool TavilySearchResults(max_results3) # 使用Tavily搜索工具限制3条结果 llm ChatOpenAI(modelgpt-4o-mini) # 使用一个性价比高的模型 # 节点1搜索节点 def search_node(state: AgentState) - AgentState: 根据主题执行网络搜索 print(f[搜索节点] 正在搜索主题: {state[topic]}) try: results search_tool.invoke(state[topic]) # Tavily返回的结果是一个字典列表我们提取‘content’字段 contents [result.get(content, ) for result in results] state[raw_search_results] contents state[iteration_count] 1 print(f[搜索节点] 搜索完成找到 {len(contents)} 条结果。) except Exception as e: print(f[搜索节点] 搜索出错: {e}) state[raw_search_results] [] return state # 节点2总结节点 def summarize_node(state: AgentState) - AgentState: 总结搜索到的原始内容 if not state[raw_search_results]: state[summarized_points] [未找到相关信息。] return state print(f[总结节点] 正在总结 {len(state[raw_search_results])} 条信息...) # 构建一个提示词模板让LLM进行总结 prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的研究助理。请将以下关于某一主题的搜索内容提炼成3-5个最核心的观点或事实。保持简洁、客观。), (user, 主题{topic}\n\n搜索内容\n{content}) ]) summarized [] # 简单起见这里我们将所有内容拼接后一次性总结。对于大量内容更佳实践是分块总结再聚合。 combined_content \n---\n.join(state[raw_search_results]) chain prompt | llm response chain.invoke({topic: state[topic], content: combined_content}) summarized.append(response.content) state[summarized_points] summarized state[iteration_count] 1 print(f[总结节点] 总结完成。) return state # 节点3报告生成节点 def report_node(state: AgentState) - AgentState: 根据总结的观点生成报告大纲 if not state[summarized_points] or state[summarized_points][0] 未找到相关信息。: state[report_outline] 无法生成报告信息不足。 return state print(f[报告节点] 正在生成报告大纲...) prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深研究员。请根据提供的核心观点为一份研究报告生成一个结构清晰的大纲。大纲应包含引言、主要章节至少3章和结论。), (user, 研究主题{topic}\n核心观点\n{points}) ]) points_text \n.join(f- {p} for p in state[summarized_points]) chain prompt | llm response chain.invoke({topic: state[topic], points: points_text}) state[report_outline] response.content state[iteration_count] 1 print(f[报告节点] 报告大纲生成完成。) return state4.2 构图与条件边让智能体学会“判断”现在我们将节点组装成图并定义它们之间的流转逻辑。我们想让智能体更智能一点如果搜索节点没有找到任何结果就直接结束跳过总结和报告。from langgraph.graph import StateGraph, END # 创建一个图并指定状态的结构 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(generate_report, report_node) # 设置入口点 workflow.set_entry_point(search) # 定义条件边在搜索节点之后判断是否有结果 def should_continue(state: AgentState) - str: 根据搜索结果决定下一步 if state[raw_search_results]: # 如果有搜索结果 return summarize # 前往总结节点 else: return END # 直接结束 # 添加从“search”出发的条件边 workflow.add_conditional_edges( search, should_continue, # 条件判断函数 { summarize: summarize, # 如果返回“summarize”则跳转到summarize节点 END: END # 如果返回END则结束 } ) # 添加普通边从总结节点到报告节点是固定的 workflow.add_edge(summarize, generate_report) # 报告节点是终点 workflow.add_edge(generate_report, END) # 编译图得到可执行的应用 app workflow.compile()4.3 运行与可视化现在我们可以运行这个智能体并查看其内部状态的变化。# 定义初始状态 initial_state: AgentState { topic: 大型语言模型在医疗诊断中的最新应用进展, raw_search_results: [], summarized_points: [], report_outline: , iteration_count: 0 } # 运行智能体 print(*50) print(开始执行研究助手智能体...) print(*50) final_state app.invoke(initial_state) print(\n *50) print(执行完成最终状态) print(*50) print(f主题: {final_state[topic]}) print(f迭代计数: {final_state[iteration_count]}) print(f\n生成的报告大纲:\n{final_state[report_outline]})为了更直观地理解执行过程LangGraph提供了强大的可视化功能。# 将图导出为PNG图片需要安装graphviz try: from IPython.display import Image, display # 获取图的PNG数据 png_data app.get_graph().draw_mermaid_png() display(Image(png_data)) except ImportError: # 如果不方便显示图片可以打印文本表示 print(app.get_graph().draw_mermaid())这张图会清晰地展示出三个节点search, summarize, generate_report以及从search出发的两条条件边一条指向summarize一条指向END。这就是你智能体的“大脑结构图”。5. 进阶技巧记忆、工具与多智能体协作构建了基础智能体后我们可以探索一些更高级的特性让智能体变得更强大、更实用。5.1 为智能体注入“记忆”上面的例子中状态是临时的。但在真实对话场景中我们需要智能体记住之前的交互。LangGraph通过Checkpointer来实现持久化状态。from langgraph.checkpoint.memory import MemorySaver # 创建一个内存检查点管理器生产环境可用数据库后端 memory MemorySaver() # 在编译图时传入检查点管理器 app_with_memory workflow.compile(checkpointermemory) # 使用一个线程ID来模拟一次对话会话 config {configurable: {thread_id: user_123_session_1}} # 第一次调用 initial_state {topic: 什么是碳中和, ...} # 省略其他字段 result1 app_with_memory.invoke(initial_state, config) print(f第一次调用后报告大纲长度: {len(result1[report_outline])}) # 基于同一thread_id再次调用可以更新主题继续研究状态会继承 new_state_for_same_thread {topic: 它与碳达峰有什么区别} # 注意这里传入的是状态的“更新部分”LangGraph会与上次检查点的状态合并 result2 app_with_memory.invoke(new_state_for_same_thread, config) print(f第二次调用后迭代计数累加: {result2[iteration_count]})通过Checkpointer智能体的状态包括中间结果可以被保存和恢复这使得构建多轮对话、长时运行任务成为可能。5.2 集成更丰富的工具智能体的能力取决于它可用的工具。LangChain-Community库提供了海量工具从搜索引擎、数据库查问到代码执行、文件操作。from langchain_community.tools import WikipediaQueryRun, DuckDuckGoSearchRun from langchain_community.utilities import WikipediaAPIWrapper, DuckDuckGoSearchAPIWrapper # 集成多个工具 wikipedia WikipediaQueryRun(api_wrapperWikipediaAPIWrapper()) duckduckgo DuckDuckGoSearchRun(api_wrapperDuckDuckGoSearchAPIWrapper()) # 你可以修改search_node让它尝试多个工具或者让一个“路由Agent”来决定使用哪个工具。 # 更高级的用法是使用LangChain的ToolExecutor和多工具Agent。5.3 构建多智能体协作系统Supervisor这是LangGraph非常强大的一个特性。你可以创建多个各司其职的智能体例如一个“研究员”一个“写手”一个“校对员”然后用一个“主管”智能体来协调它们的工作。from langgraph.graph import MessagesState from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI # 1. 创建不同的智能体每个智能体其实是一个LangGraph子图 llm ChatOpenAI(modelgpt-4o-mini) # 研究员智能体擅长搜索和总结 research_agent create_react_agent(llm, tools[search_tool, wikipedia]) # 写手智能体擅长结构化写作 writer_agent create_react_agent(llm, tools[]) # 可以给写手一些写作辅助工具 # 校对员智能体擅长批判性审查 critic_agent create_react_agent(llm, tools[]) # 2. 定义一个主管智能体它的工作是分配任务 supervisor_node create_react_agent( llm, tools[], # 主管不直接使用工具而是调用其他智能体 # 需要给主管一个特殊的提示词告诉它手下有哪些成员以及他们的职责 prompt你是主管负责协调研究员、写手和校对员来完成报告。根据用户请求和当前进展决定下一步该让谁工作或者任务是否完成。 ) # 3. 构建一个更大的图包含主管节点和各个工作节点并定义他们之间的消息传递逻辑。 # 这涉及到更复杂的图构建add_node, add_conditional_edges用于路由消息。 # 官方示例‘Multi-Agent Collaboration’提供了完整模板。这种架构非常适合复杂任务分解每个子智能体可以专注于自己擅长的领域由主管进行全局调度从而产生更高质量、更可靠的结果。6. 生产环境部署与性能优化考量当你的智能体原型在本地运行良好后下一步就是考虑如何将它部署为稳定的服务。6.1 部署模式选择API服务FastAPI / Flask这是最常见的方式。将编译好的LangGraphapp封装成FastAPI的端点。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import asyncio app_fastapi FastAPI() # 假设 app 是你编译好的LangGraph应用 # langgraph_app workflow.compile() class AgentRequest(BaseModel): topic: str thread_id: str None app_fastapi.post(/research) async def run_research(request: AgentRequest): try: config {configurable: {thread_id: request.thread_id or default_thread}} initial_state {topic: request.topic, ...} # LangGraph的invoke是同步的在异步环境中需在线程池中运行 result await asyncio.to_thread(app.invoke, initial_state, config) return {report: result[report_outline], status: success} except Exception as e: raise HTTPException(status_code500, detailstr(e))使用uvicorn等ASGI服务器运行即可。异步流式响应对于耗时长的工作流提供流式接口Server-Sent Events能极大提升用户体验让前端实时看到“思考过程”。from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph workflow StateGraph(...) # ... 构建图 ... app workflow.compile(checkpointerMemorySaver()) # 流式调用 async def stream_research(topic: str, thread_id: str): config {configurable: {thread_id: thread_id}} initial_state {topic: topic, ...} async for event in app.astream_events(initial_state, config, versionv2): kind event[event] if kind on_chain_start: if event[name] search_node: yield fdata: 开始搜索...\n\n elif kind on_chain_end: if event[name] generate_report: yield fdata: 报告生成完成\n\n yield fdata: {event[data][output][report_outline]}\n\n前端通过EventSource连接这个端点就能收到实时更新。Serverless函数对于轻量级、偶发性的任务可以部署为Vercel、AWS Lambda或Google Cloud Function。需要特别注意冷启动时间以及依赖包的大小Python层可能需要精简依赖TypeScript层更有优势。6.2 性能与成本优化LLM调用优化缓存对相同的提示词进行缓存可以大幅减少费用和延迟。LangChain集成了一些缓存方案如InMemoryCache,SQLiteCache。模型选型并非所有任务都需要GPT-4。对于搜索总结、文本格式化等任务gpt-3.5-turbo或gpt-4o-mini可能更具性价比。对于路由、分类等简单任务甚至可以使用更小、更快的模型。批处理如果有多条独立内容需要总结或翻译可以将它们组合在一个提示词中批量请求LLM而不是多次调用。图执行优化设置超时与重试为外部工具调用如搜索API设置合理的超时和重试机制避免单个节点失败导致整个工作流卡死。异步节点如果节点任务主要是I/O密集型如调用多个独立的API可以将其定义为异步函数并在图中使用异步执行提升整体吞吐量。简化状态保持State结构尽量精简只存放必要数据。过大的状态会在节点间序列化/反序列化时带来开销。监控与可观测性日志记录在每个关键节点记录输入、输出和耗时。可以使用langsmithLangChain官方平台进行完整的跟踪和评估。设置断点利用LangGraph的interrupt机制在关键节点如生成最终报告前暂停引入人工审核确保输出质量符合要求。7. 常见陷阱、排查指南与最佳实践在开发过程中我踩过不少坑这里总结出来希望能帮你绕过去。7.1 常见问题与解决方案问题现象可能原因排查步骤与解决方案图编译或执行时报KeyError状态State字典的键在节点中访问不存在。1. 检查TypedDict的定义是否与节点中实际访问的键完全一致。2. 确保每个节点返回的字典都包含所有在TypedDict中定义的键即使值为空列表或None。条件边add_conditional_edges不生效条件判断函数返回的值与add_conditional_edges中映射的键不匹配。1. 打印条件函数的返回值确认它是summarize、END还是其他你定义的字符串。2. 检查add_conditional_edges的第三个参数字典其键必须与条件函数所有可能的返回值完全对应。LLM调用缓慢或超时网络问题、API限流、或提示词过于复杂导致模型响应慢。1. 在节点中添加超时设置chain prompt | llm.with_config({run_name: summarize, max_concurrency: 1})。2. 简化提示词移除不必要的指令。3. 考虑使用更快的模型或检查API密钥的速率限制。工具调用失败API密钥未设置、工具参数格式错误、或第三方服务不可用。1. 确认环境变量如TAVILY_API_KEY已正确设置。2. 在节点内部用try...except包裹工具调用并妥善处理异常更新状态避免工作流崩溃。3. 单独测试工具是否能正常工作。状态更新不符合预期如计数器没累加对Annotated字段的操作方式错误。1. 对于标记为operator.add的字段在节点中应使用state[“count”] 1而不是state[“count”] state[“count”] 1吗实际上两者在Python中效果相同但确保你是在更新字典而不是重新赋值一个全新的字典关键是要返回修改后的整个state。2. 更稳妥的方式是使用state.update({“iteration_count”: state[“iteration_count”] 1})。流式输出不工作使用的astream或astream_events版本不对或节点函数不是异步的。1. 确认使用app.astream_events(..., version“v2”)这是最新的稳定流式API。2. 如果自定义节点中有异步操作如调用异步LLM客户端确保节点函数本身是async def并且在图中正确配置。7.2 最佳实践心得从简单开始逐步迭代不要一开始就设计一个包含10个节点、复杂循环的超级智能体。先构建一个最小可行产品MVP例如只有“搜索-总结”两个节点的链确保它能跑通。然后逐步添加新节点和逻辑。可视化是你的朋友在添加每个节点和边之后都用app.get_graph().draw_mermaid()看看图的结构是否符合预期。这能帮你快速发现逻辑错误。为状态设计严谨的模式花时间好好设计TypedDict。明确的类型提示不仅能帮助IDE自动补全和查错更是你对智能体数据流的蓝图。考虑使用Pydantic模型来获得更强大的验证能力。节点功能要“单一职责”每个节点最好只做一件事并且做好。例如一个节点只负责调用搜索API另一个节点只负责解析搜索结果。这提高了可测试性和可复用性。善用“中断”进行人工审核对于生产环境在最终输出或执行关键操作如发送邮件、修改数据库前通过interrupt机制加入人工审核节点是控制风险的必要手段。为你的图编写测试像测试普通函数一样测试你的智能体。准备一些标准的输入状态调用app.invoke()然后断言输出的状态是否符合预期。这对于保证智能体逻辑的稳定性至关重要。TypeScript项目的额外提示充分利用TypeScript的类型系统。为你的State定义清晰的interface。使用langgraph的StateGraph时泛型参数能提供完美的类型安全让你在编码时就能发现状态访问的错误。开发AI智能体是一个充满探索和迭代的过程。LangChain和LangGraph提供的这套范式极大地降低了构建复杂、可靠智能体的门槛。从今天开始选择一个你感兴趣的小问题用这个框架尝试解决它。你会发现开启你的Agent时代并没有想象中那么遥远。
返回列表