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

资讯详情

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

基于LangGraph与LangChain构建复杂AI智能体:从原理到实践

基于LangGraph与LangChain构建复杂AI智能体:从原理到实践 1. 项目概述为什么我们需要一个“智能体”框架如果你最近在捣鼓大语言模型LLM应用尤其是想让它不只是个“聊天机器人”而是能真正帮你处理复杂任务、调用工具、甚至自主决策的“智能体”Agent那你大概率已经听过 LangChain、LangGraph 这些名字了。Deep Agents 框架正是在这个背景下一个值得你投入时间研究的、更聚焦于构建复杂、可扩展智能体系统的工具集或方法论。它不是某个单一的、有官方文档的库而更像是一个社区实践和高级模式的集合其核心思想是如何让多个 AI 智能体协同工作处理那些单个智能体或简单链式调用搞不定的问题。简单来说当你的需求从“问一句答一句”升级到“给我分析这份财报总结要点预测趋势并生成一份投资建议报告”时你就需要智能体框架了。一个智能体不仅仅是调用一次模型它包含状态管理、工具调用、逻辑路由、记忆存储等一系列复杂机制。Deep Agents 这个概念往往代表着对智能体能力的深度挖掘比如实现长期记忆、子任务分解、多智能体协作、以及稳定可靠的任务执行。它解决的痛点非常明确将散乱、脆弱的 Prompt 工程和工具调用代码转化为结构化、可维护、可观测的智能体系统。无论是想开发一个能自动处理工单的客服助手还是一个能联网搜索、分析数据并撰写周报的个人助理Deep Agents 框架所代表的技术栈都是你必须掌握的。2. 核心概念与生态定位LangChain、LangGraph 与 Agent在深入 Deep Agents 的具体实现之前我们必须先理清它赖以生存的生态系统。很多人会把 LangChain、LangGraph 和 Agent 这几个词混用但它们其实指代不同层次的概念。2.1 LangChain应用构建的“脚手架”你可以把 LangChain 理解为智能体世界的“Spring Boot”。它提供了一整套标准化的组件和接口让你能像搭积木一样构建 LLM 应用。其核心价值在于“连接”模型抽象层无论你用 OpenAI 的 GPT还是 Anthropic 的 Claude或是本地部署的 LlamaLangChain 提供了统一的接口切换模型可能只需要改一行配置。核心概念PromptTemplate提示词模板、Chain链将多个组件顺序执行、Memory记忆用于保存对话历史或状态、Document Loader文档加载器等。工具集成内置了大量现成的工具Tools如搜索引擎、计算器、Python REPL也允许你轻松自定义工具。在 LangChain 的早期版本中Agent是其一个重要组件它基于Chain构建但引入了“思考-行动-观察”的循环。然而随着智能体逻辑越来越复杂单纯用Chain来表述这种带有循环、分支和状态的结构显得力不从心。2.2 LangGraph智能体工作流的“编排引擎”这就是 LangGraph 登场的原因。如果说 LangChain 提供了砖块和水泥那么 LangGraph 提供了建筑蓝图和施工流程。它本质上是一个基于图Graph的工作流编排库专门为构建有状态、多步骤的智能体而设计。核心是“图”你将智能体的每个功能单元如“理解用户意图”、“调用搜索工具”、“总结信息”定义为一个节点Node节点之间的连线Edge定义了执行流程。这个流程不再是简单的链条它可以包含条件分支if-else、循环loop甚至并行执行。状态管理这是 LangGraph 的杀手锏。它引入了一个共享的State对象在整个图执行过程中流转和更新。每个节点读取状态执行操作并修改状态。这完美契合了智能体需要记忆上下文、维护中间结果的需求。与 LangChain 的关系LangGraph 可以完全独立使用但它与 LangChain 生态无缝集成。你可以直接使用 LangChain 的工具、模型和提示词模板作为 LangGraph 图中的节点。一个常见的理解是用 LangGraph 来构建和编排复杂的智能体用 LangChain 的组件来填充这个智能体的具体能力。2.3 Agent从概念到实现最后来说Agent。在 Deep Agents 的语境下它指的是一个完整的、可执行的智能实体。一个Agent通常由以下部分构成一个大脑LLM负责理解和规划。一套工具Tools用于执行具体动作如搜索、写文件、执行代码。一个记忆系统Memory短期记忆当前会话上下文和长期记忆向量数据库等。一个执行引擎Orchestrator这就是 LangGraph 扮演的角色它决定了大脑在什么情况下使用什么工具并管理整个执行流程。因此Deep Agents 框架通常指的是以 LangGraph 为核心编排层深度融合 LangChain 生态组件旨在构建具备深度推理和复杂行动能力的智能体系统的最佳实践和设计模式。注意市面上也有一些其他优秀的 Agent 框架如 Dify、CrewAI 等。Dify 更偏向低代码/可视化平台开箱即用但定制性相对受限CrewAI 专注于多智能体协作。而 LangChain LangGraph 的组合提供了从底层到高层的最大灵活性和控制力是“从入门到精通”路径上必须征服的领域。3. 环境搭建与核心组件初始化理论说再多不如动手搭一个。我们从一个经典的“研究助手”智能体开始它需要能联网搜索、总结信息并保存结果。假设你已有 Python 环境我们一步步来。3.1 基础环境与安装首先创建虚拟环境并安装核心库。这里我们选择langchain和langgraph同时为了联网搜索安装langchain-community社区工具集和tavily-python一个不错的搜索 API。# 创建并激活虚拟环境以 conda 为例 conda create -n deep-agents python3.11 conda activate deep-agents # 安装核心库 pip install langchain langgraph langchain-community # 安装搜索工具和 OpenAI 库假设使用 GPT 模型 pip install tavily-python openai # 可选用于结构化输出的库能让模型输出更规范的 JSON pip install langchain-openai3.2 配置密钥与模型在项目根目录创建一个.env文件来管理密钥永远不要将密钥硬编码在代码中。# .env 文件内容 OPENAI_API_KEYsk-your-openai-key-here TAVILY_API_KEYyour-tavily-key-here然后在代码中加载环境变量并初始化模型。这里我们使用ChatOpenAI并指定gpt-4-turbo或gpt-3.5-turbo后者成本更低适合实验。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 加载环境变量 load_dotenv() # 初始化 LLM llm ChatOpenAI( modelgpt-4-turbo, # 或 gpt-3.5-turbo temperature0, # 对于任务执行类Agent低温度0-0.2输出更稳定 api_keyos.getenv(OPENAI_API_KEY) )3.3 构建你的第一个工具Tool工具是智能体的手和脚。我们创建一个使用 Tavily 进行搜索的工具。from langchain_community.tools.tavily_search import TavilySearchResults # 初始化搜索工具 search_tool TavilySearchResults( api_keyos.getenv(TAVILY_API_KEY), max_results3 # 每次搜索返回3条结果避免信息过载 ) # 为了演示我们再创建一个简单的“文本写入文件”的工具 from langchain.tools import tool import datetime tool def write_to_file(content: str, filename: str None): 将内容写入文件。如果未提供文件名则使用时间戳生成。 if filename is None: filename fresearch_{datetime.datetime.now().strftime(%Y%m%d_%H%M%S)}.md with open(filename, w, encodingutf-8) as f: f.write(content) return f内容已成功写入文件{filename} # 将工具打包成列表后续供智能体使用 tools [search_tool, write_to_file]实操心得在定义自定义工具时务必使用tool装饰器并为函数编写清晰、准确的文档字符串docstring。LLM 会阅读这些文档字符串来理解工具的用途和参数这是工具能否被正确调用的关键。参数类型提示如content: str也能帮助 LangChain 进行更好的类型校验和转换。4. 构建智能体工作流从 State 设计到 Graph 编排这是 Deep Agents 框架最核心的部分。我们将使用 LangGraph 来构建一个具备“搜索-分析-存档”流程的智能体。4.1 定义状态State状态是智能体的“记忆画布”它定义了在整个工作流中流转的数据结构。我们使用TypedDict来定义。from typing import TypedDict, Annotated, List, Union from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史LangGraph 内置的 add_messages 函数会处理它 messages: Annotated[List, add_messages] # 用户原始查询 user_query: str # 从网络搜索到的原始内容 raw_search_results: List[str] # 经过分析总结后的最终内容 analyzed_content: str # 一个标志位指示是否应该继续循环例如是否需要进一步搜索 should_continue: boolAnnotated和add_messages是 LangGraph 的语法糖用于自动将新的消息追加到messages列表中这对于维护对话历史非常方便。4.2 创建节点Nodes节点是图中的一个步骤。我们将创建三个核心节点理解与规划、执行搜索、分析与总结。from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage from langgraph.prebuilt import ToolExecutor # 首先创建一个 ToolExecutor 来管理工具调用 tool_executor ToolExecutor(tools) # 节点1理解用户意图并规划步骤 def understand_and_plan(state: AgentState): print(f[节点理解与规划] 正在处理查询{state[user_query]}) # 构建系统提示词指导模型进行规划 system_prompt 你是一个研究助手。你的任务是分析用户的问题并决定是否需要通过网络搜索来获取最新信息。 如果用户的问题涉及实时信息、最新事件、具体数据或你不知道的知识你就需要搜索。 你只需要回答“NEED_SEARCH”或“DIRECT_ANSWER”。如果回答“NEED_SEARCH”请同时生成一个用于搜索的查询词。 messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentstate[user_query]) ] # 调用 LLM 进行判断 response llm.invoke(messages) decision response.content.strip() new_state {**state} # 复制原状态 if NEED_SEARCH in decision: # 简单地从决策文本中提取搜索词实际生产环境可用更复杂的解析如 Pydantic 输出 # 假设模型返回格式为NEED_SEARCH: 查询词 if : in decision: search_query decision.split(:, 1)[1].strip() else: # 如果模型没按格式来用原问题作为搜索词后备方案 search_query state[user_query] new_state[should_continue] True # 将搜索词存入状态供下一个节点使用 new_state[search_query] search_query print(f 决策需要搜索。搜索词{search_query}) else: new_state[should_continue] False new_state[analyzed_content] 根据我的知识无需搜索即可回答... # 此处简化实际应调用LLM生成答案 print.f( 决策直接回答。) return new_state # 节点2执行搜索工具 def execute_search(state: AgentState): if not state.get(should_continue): return state # 如果不需要搜索直接跳过 search_query state.get(search_query, state[user_query]) print(f[节点执行搜索] 正在搜索{search_query}) try: # 调用搜索工具 search_results search_tool.invoke({query: search_query}) # Tavily 返回的结果是一个列表每个元素是字典。我们提取片段snippet。 snippets [result.get(content, )[:500] for result in search_results] # 截取前500字符 new_state {**state, raw_search_results: snippets} print(f 搜索完成获取到 {len(snippets)} 条结果摘要。) return new_state except Exception as e: print(f 搜索失败{e}) # 即使搜索失败也更新状态避免流程卡住 new_state {**state, raw_search_results: [f搜索出错{str(e)}]} return new_state # 节点3分析结果并总结 def analyze_and_summarize(state: AgentState): if not state.get(raw_search_results): # 如果没有搜索结果则直接返回之前生成的直接答案或提示 if state.get(analyzed_content): return state else: new_state {**state, analyzed_content: 未能获取到相关信息。} return new_state print(f[节点分析总结] 正在分析 {len(state[raw_search_results])} 条搜索结果。) # 构建提示词让模型基于搜索结果进行总结 search_context \n\n---\n\n.join(state[raw_search_results]) prompt f请基于以下搜索到的信息回答用户的原始问题。 用户问题{state[user_query]} 搜索到的信息 {search_context} 请生成一份结构清晰、信息准确的回答。如果信息不足或矛盾请明确指出。 messages [HumanMessage(contentprompt)] response llm.invoke(messages) new_state {**state, analyzed_content: response.content} print(f 分析总结完成。) return new_state4.3 定义条件边Conditional Edges与构图现在我们需要将这些节点连接起来并决定流程如何流转。这里的关键是理解与规划节点后的条件判断。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(understand_and_plan, understand_and_plan) workflow.add_node(execute_search, execute_search) workflow.add_node(analyze_and_summarize, analyze_and_summarize) # 设置入口点 workflow.set_entry_point(understand_and_plan) # 定义条件路由函数 def route_after_planning(state: AgentState): 根据规划节点的结果决定下一步是去搜索还是直接结束。 if state.get(should_continue): return execute_search else: return END # 直接结束流程 # 添加从“理解与规划”节点出发的边这是一个条件边 workflow.add_conditional_edges( understand_and_plan, route_after_planning, { execute_search: execute_search, END: END } ) # 添加从“执行搜索”到“分析总结”的普通边 workflow.add_edge(execute_search, analyze_and_summarize) # 添加从“分析总结”到结束的边 workflow.add_edge(analyze_and_summarize, END) # 编译图 app workflow.compile()4.4 运行你的第一个智能体图编译完成后就可以运行了。# 定义初始状态 initial_state: AgentState { messages: [], # 初始消息为空 user_query: 2024年人工智能领域有哪些重要的趋势, raw_search_results: [], analyzed_content: , should_continue: True # 初始设为True由第一个节点判断 } # 运行图 try: final_state app.invoke(initial_state) print(\n *50) print(智能体执行完成) print(*50) print(f最终总结内容\n{final_state[analyzed_content][:500]}...) # 打印前500字符 except Exception as e: print(f执行过程中出错{e})运行这段代码你会在控制台看到智能体一步步执行理解问题、判断需要搜索、执行搜索、分析结果并生成总结。一个最基本的 Deep Agent 就诞生了。注意事项这个示例为了清晰将逻辑拆分得很细。在实际项目中understand_and_plan节点往往会被更强大的“代理节点”取代这个代理节点能自动选择并调用工具我们将在下一章深入。5. 高级模式工具调用代理、记忆与多智能体协作基础工作流只能算入门。Deep Agents 的威力体现在其高级模式上。5.1 使用 LangChain 的 Agent Executor 作为节点手动判断该调用哪个工具如我们之前的understand_and_plan节点是低效的。更好的方式是让 LLM 自己决定。我们可以用 LangChain 的create_react_agent或create_tool_calling_agent来构建一个能自动选择工具的“代理节点”。from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate # 1. 定义提示词模板 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个强大的研究助手。你可以使用工具来获取实时信息。 请用中文回答用户的问题。如果你需要更多信息请使用搜索工具。 在得到足够信息后你可以选择将最终结果写入文件。), (placeholder, {messages}), # 这里会自动填充消息历史 ]) # 2. 创建 Agent agent_runnable create_tool_calling_agent(llm, tools, prompt_template) # 3. 创建 AgentExecutor它封装了“思考-调用工具-观察”的循环 agent_node AgentExecutor(agentagent_runnable, toolstools, handle_parsing_errorsTrue) # 现在我们可以在 LangGraph 中创建一个使用这个代理的节点 def advanced_agent_node(state: AgentState): 一个集成了工具调用能力的智能节点 # 从状态中构建输入。AgentExecutor 期望的输入格式是包含“input”和“chat_history”的字典。 # 我们需要把 state[messages] 转换成合适的格式。 human_input state[user_query] chat_history state.get(messages, []) # 简单处理将历史消息和当前问题组合成输入 input_for_agent { input: human_input, # AgentExecutor 通常能处理消息列表格式的历史 chat_history: chat_history[-5:], # 只保留最近5轮历史防止上下文过长 } print(f[高级代理节点] 开始处理{human_input}) result agent_node.invoke(input_for_agent) # 更新状态。result[output] 包含代理的最终回答。 new_messages state[messages] [HumanMessage(contenthuman_input), AIMessage(contentresult[output])] new_state { **state, messages: new_messages, analyzed_content: result[output] # 将最终输出也存到 analyzed_content } return new_state将这个advanced_agent_node添加到图中它可以独立完成“接收问题 - 思考是否需要工具 - 调用工具 - 整合答案”的全过程比我们之前手动拆分的三个节点更强大和简洁。5.2 集成记忆系统短期记忆对话历史我们已经通过state[‘messages’]实现了。长期记忆如向量数据库可以让智能体记住跨会话的信息。以下是一个集成 Chroma 向量数据库的简化示例from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document # 初始化嵌入模型和向量数据库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def store_memory_node(state: AgentState): 将本次交互的重要信息存入长期记忆 if not state.get(analyzed_content): return state print([记忆节点] 正在存储到长期记忆。) # 1. 将总结的内容创建为文档 doc Document(page_contentstate[analyzed_content], metadata{query: state[user_query], timestamp: datetime.datetime.now().isoformat()}) # 2. 分割文本如果内容很长 text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200) splits text_splitter.split_documents([doc]) # 3. 存入向量库 vectorstore.add_documents(splits) return state def retrieve_memory_node(state: AgentState): 从长期记忆中检索与当前问题相关的历史信息 query state[user_query] print(f[记忆节点] 正在从长期记忆中检索{query}) # 检索最相关的3个片段 docs vectorstore.similarity_search(query, k3) retrieved_context \n.join([doc.page_content for doc in docs]) # 将检索到的上下文添加到用户查询中作为增强的输入 enhanced_query f基于以下历史记录和当前问题请回答问题。 相关历史记录 {retrieved_context} 当前问题{query} new_state {**state, user_query: enhanced_query} return new_state你可以在工作流的开始添加retrieve_memory_node在结束添加store_memory_node这样智能体就具备了基于上下文的记忆能力。5.3 多智能体协作模式对于极其复杂的任务可以设计多个各司其职的智能体协同工作。例如一个“研究项目”可以由以下智能体组成主管Supervisor接收用户指令将任务分解分配给其他智能体并整合最终结果。研究员Researcher专门负责搜索和收集信息。分析师Analyst负责对收集的信息进行深度分析和总结。撰稿人Writer负责将分析结果格式化为漂亮的报告。在 LangGraph 中这可以通过创建多个子图每个子图代表一个智能体并由一个主图主管来协调它们实现。这涉及到更复杂的State设计需要包含任务列表、分配状态、中间结果池等和节点间的消息传递。这是 Deep Agents 框架的进阶课题其核心思想是将复杂问题分解通过智能体间的通信通常通过更新共享的State来共同解决。6. 调试、监控与生产化部署构建一个能跑的智能体只是第一步让它稳定、可靠、可观测地运行才是真正的挑战。6.1 可视化与调试LangGraph 提供了内置的可视化功能对于理解工作流至关重要。# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以打印文本表示 print(app.get_graph().draw_mermaid())在调试时充分利用print语句如我们之前在各节点中添加的来跟踪状态变化。对于更复杂的调试可以修改节点函数让其将关键步骤的输入输出记录到日志文件或像LangSmith这样的专业平台。6.2 错误处理与稳定性智能体工作流可能因为多种原因失败工具 API 调用超时、LLM 返回格式错误、网络问题等。必须在关键节点添加健壮的错误处理。重试机制对于工具调用如搜索、API请求使用tenacity库实现指数退避重试。Fallback 策略如果一个工具调用失败是否有备用工具或备用数据源超时控制为每个节点或整个图的执行设置超时时间防止无限期卡住。状态检查在节点开始执行前检查输入状态是否包含必需的字段格式是否正确。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def robust_search_tool_call(query): 带有重试机制的搜索工具调用 return search_tool.invoke({query: query}) def safe_execute_search(state: AgentState): try: # ... 使用 robust_search_tool_call ... except Exception as e: # 记录错误并更新状态以反映失败让流程可以继续或优雅结束 state[raw_search_results] [fError: {str(e)}] state[should_continue] False # 或者触发一个错误处理节点 return state6.3 生产化考量配置管理将所有配置模型参数、API密钥、工具参数外置到配置文件或环境变量。异步执行LangGraph 支持异步ainvoke对于高并发场景使用异步可以大幅提高吞吐量。持久化状态对于长时间运行或需要中断恢复的工作流需要将State序列化后存储到数据库如 Redis、PostgreSQL。可观测性集成像LangSmith这样的平台它可以追踪每一次 LLM 调用、工具调用和图的执行路径记录输入输出、延迟和成本是调试和优化不可或缺的工具。安全性对用户输入进行严格的清洗和校验防止 Prompt 注入攻击。对工具调用进行权限控制特别是执行代码、读写文件等危险操作的工具。7. 常见问题与排查技巧实录在实际开发和运行 Deep Agents 应用时你会遇到各种各样的问题。这里记录一些典型问题和解决思路。7.1 工具调用失败或不被识别问题智能体输出类似“我需要搜索一下”但并未实际调用搜索工具或者调用工具时参数错误。排查检查工具定义确保自定义工具的tool装饰器使用正确且文档字符串清晰描述了功能和参数。可以单独测试工具是否能被正常调用search_tool.invoke({“query”: “test”})。检查提示词给 Agent 的系统提示词System Prompt必须明确告知它有哪些工具可用以及何时使用。可以在提示词中列出工具名称和描述。检查模型能力gpt-3.5-turbo的工具调用能力弱于gpt-4-turbo。如果使用前者遇到复杂工具调用时可能不稳定升级模型是首选方案。启用详细日志在初始化AgentExecutor时设置verboseTrue可以打印出模型决定调用工具时的思考过程帮助你诊断问题。7.2 状态State管理混乱问题节点之间传递的数据丢失或格式不对导致后续节点报KeyError。排查明确定义 State 结构使用TypedDict严格定义每个字段的类型。这不仅是类型提示也是设计文档。每个节点只读写特定的字段遵循“单一职责”一个节点只修改它负责的字段避免随意修改全局状态。打印调试在每个节点的开始和结束打印出传入的state和即将返回的new_state的关键字段这是最直接的调试方法。使用 Pydantic 进行验证可以将State定义为一个 PydanticBaseModelLangGraph 支持 Pydantic State它能在节点间自动进行数据验证和序列化非常强大。7.3 工作流陷入无限循环问题智能体反复执行同一个操作无法跳出循环。排查检查条件边Conditional Edges逻辑确保你的路由函数如route_after_planning在所有可能的分支下都能正确返回一个节点名或END。逻辑漏洞可能导致它在几个节点间来回跳转。设置最大迭代次数在编译图时可以使用checkpointer或自定义逻辑来限制循环次数。AgentExecutor内部就有max_iterations参数来防止死循环。审查工具输出有时工具返回的结果会让 LLM 认为任务未完成从而再次调用同一个工具。检查工具返回的内容是否清晰、完整能否让 LLM 做出“任务完成”的判断。7.4 性能与成本优化问题响应速度慢API 调用费用高。优化缓存对频繁且结果不变的 LLM 调用或工具调用如查询静态知识实施缓存。LangChain 提供了InMemoryCache、RedisCache等。精简上下文严格控制传入 LLM 的上下文长度。只保留最相关的对话历史、工具结果和记忆片段。使用RecursiveCharacterTextSplitter合理分割长文档。模型选型在非核心推理环节使用更小、更快的模型如gpt-3.5-turbo在需要深度分析时再切换到大模型。异步并行如果工作流中有多个独立的任务考虑使用 LangGraph 的StateGraph的并发能力或者用asyncio.gather在自定义节点中并行执行。7.5 与前端或其他服务集成问题如何将 LangGraph 智能体作为 API 提供服务方案使用 FastAPI将编译好的app对象封装在 FastAPI 端点中。因为app.invoke(state)通常是同步的在高并发下可能阻塞建议使用线程池或将其改造成异步端点。from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel import asyncio from concurrent.futures import ThreadPoolExecutor app_fastapi FastAPI() executor ThreadPoolExecutor(max_workers4) class QueryRequest(BaseModel): question: str app_fastapi.post(/ask) async def ask_agent(request: QueryRequest): loop asyncio.get_event_loop() initial_state {...} # 根据 request 构建初始状态 # 在线程池中运行同步的 invoke 调用 final_state await loop.run_in_executor(executor, app.invoke, initial_state) return {answer: final_state[analyzed_content]}流式输出Streaming对于生成时间较长的回答可以使用 LangChain 和 LangGraph 的流式支持将 token 逐个返回给前端提升用户体验。这需要前后端都支持 Server-Sent Events (SSE) 或 WebSocket。从搭建一个简单的搜索总结智能体到设计具备记忆和协作能力的复杂多智能体系统再到考虑生产环境的稳定性、可观测性和集成这条路径涵盖了 Deep Agents 框架从入门到精通的核心。最关键的是动手实践从一个具体的小需求开始逐步增加复杂度在踩坑和解决问题的过程中你会对智能体的本质有更深刻的理解。这个领域迭代飞快但掌握 LangChain 和 LangGraph 这套以“图”和“状态”为核心的设计范式能让你从容应对各种新的 Agent 框架和模式。
返回列表