1. 项目概述LangSmith Fleet是什么如果你最近在AI应用开发特别是基于大语言模型LLM的Agent或复杂工作流构建领域里摸爬滚打那么“LangSmith”这个名字你一定不陌生。它早已从一个单纯的调试工具演变成了LangChain生态中事实上的“开发者控制台”。但今天要聊的不是LangSmith本身而是它的一个全新、且在我看来极具颠覆性的能力扩展——LangSmith Fleet。简单来说LangSmith Fleet是LangSmith平台推出的一个多智能体Multi-Agent协作与编排框架。它允许你将多个独立的、具备特定功能的LangChain Agent或任何可调用的LLM工作流组织成一个“舰队”Fleet让它们能够协同工作共同完成一个更复杂的任务。这不再是单个Agent的单打独斗而是一个有组织、有分工、有协作的“团队作战”模式。为什么说它重要因为现实世界的问题很少是线性的。一个客服机器人可能需要先理解用户意图分类Agent再查询知识库检索Agent最后生成个性化回复生成Agent。一个数据分析任务可能需要数据提取、清洗、分析和可视化等多个步骤每个步骤都可能需要一个专门的Agent来处理。LangSmith Fleet就是为了解决这种“任务分解与协作”的痛点而生的。它让你能够像搭积木一样将不同的AI能力模块组合起来并通过一个中央控制器Fleet来协调它们的执行顺序和数据流转。2. 核心设计思路从单兵到舰队的进化2.1 传统Agent模式的瓶颈在LangSmith Fleet出现之前我们构建复杂AI应用主要有两种思路巨型单体Agent把所有逻辑和工具调用都塞进一个Prompt里让一个LLM去“思考”所有步骤。这就像让一个CEO去同时处理战略规划、财务审计、代码开发和客户服务。结果往往是Prompt极其臃肿上下文窗口消耗巨大LLM容易“分心”或遗忘早期指令导致任务失败率高且调试起来如同大海捞针。手动编排脚本自己写Python脚本用if-else或状态机来手动调用不同的LangChain Chain或Agent。这给了你完全的控制权但代价是编排逻辑与业务逻辑高度耦合。每增加一个步骤或修改流程你都需要重写核心调度代码。更头疼的是这种分散的调用很难在LangSmith中获得一个统一的、端到端的执行轨迹视图进行调试。这两种方式都让构建可靠、可维护的复杂应用变得异常困难。2.2 Fleet的设计哲学声明式编排与集中式可观测性LangSmith Fleet的核心理念正是为了解决上述问题。它的设计哲学可以概括为两点第一声明式编排。你不再需要编写复杂的流程控制代码。相反你通过一种更高级的、声明性的方式来定义你的“舰队”成员即各个Agent或可调用对象以及它们之间的协作关系。LangSmith特别是结合LangGraph提供了直观的方式来描述工作流哪个Agent先执行在什么条件下触发下一个Agent如何传递数据。这极大地降低了编排的复杂度让开发者能更专注于每个“船员”Agent本身的能力建设。第二集中式可观测性。这是LangSmith的老本行也是Fleet价值倍增的关键。当你的多个Agent在一个Fleet中协作时LangSmith可以捕获整个工作流的完整执行轨迹。你可以在一个统一的界面上清晰地看到请求是如何在Agent A、B、C之间流转的每个步骤的输入输出是什么耗时多少调用了哪些工具LLM的思考过程如果开启如何。这种端到端的可观测性对于调试一个多步骤、可能出错的协作流程来说是无可替代的。注意虽然网络热词中提到了“langgraph langsmith langchain agent ui”这恰好点明了它们的关系。LangGraph是LangChain生态中用于构建有状态、多步骤工作流图的库。LangSmith Fleet可以看作是LangSmith平台对这类由LangGraph构建的、或多个独立Agent组成的复杂工作流提供的托管、监控与调试增强套件。UI部分则是指LangSmith提供的可视化界面让你能直观地管理Fleet和查看运行日志。2.3 一个类比从快递员到物流公司想象一下单个Agent就像一个优秀的快递员他能高效地完成“从A点取件送到B点”的任务。但如果你要处理“从全球多个仓库集货进行分拣、跨境运输、清关、最后一公里配送”的复杂物流链路一个快递员就无能为力了。这时你需要一个物流公司Fleet。这个公司有调度中心Fleet Orchestrator根据订单用户请求决定整个流程。仓储Agent负责货品管理和出库。运输Agent负责干线物流。清关Agent处理法规文件。配送Agent完成最终交付。LangSmith Fleet就是帮你搭建这个“物流公司”的框架和运营平台。它提供调度规则编排并给你一个全局监控大屏LangSmith UI让你能实时看到每个包裹在哪个环节、状态如何一旦某个环节如清关卡住了你能立刻定位并介入。3. 核心组件与架构拆解要理解并使用LangSmith Fleet我们需要先厘清几个核心概念。它们共同构成了Fleet的骨架。3.1 Fleet舰队本身一个Fleet是一个逻辑上的容器代表一个完整的、可执行的多智能体应用。它包含以下要素成员Members即组成这个舰队的各个智能体Agent或可调用单元。每个成员都有唯一的标识符。编排逻辑Orchestration Logic定义了成员之间如何协作。这可以通过LangGraph的图Graph来定义也可以通过更简单的配置来指定执行顺序和条件分支。共享上下文Shared ContextFleet内的成员如何共享数据通常一个共享的“状态”State对象会在成员间传递每个成员可以读取和修改状态中与自己相关的部分。配置与权限例如为整个Fleet设置统一的LLM模型、温度参数或者管理哪些API密钥可以被成员访问。在LangSmith UI中一个Fleet可能会被作为一个独立的项目或实体进行管理你可以查看它的整体运行指标、调用日志和成本分析。3.2 Agent成员舰队的船员这里的Agent是广义的它可以是一个标准的LangChain Agent具备工具调用能力和ReAct式推理循环。一个LangChain Chain例如一个简单的检索问答链RetrievalQA。一个普通的Python函数Callable任何可以被调用的对象用于执行确定性逻辑如数据格式化、计算。一个封装好的API端点。在Fleet的语境下每个成员都应该被设计为“单一职责”的。理想情况下一个成员只做好一件事。例如一个“网络搜索Agent”一个“代码执行Agent”一个“总结归纳Agent”。这种设计使得每个成员易于开发、测试和复用。3.3 编排引擎舰队的调度系统这是Fleet的大脑。目前LangChain生态中LangGraph是实现复杂编排的首选工具。LangGraph允许你以“图”的形式定义工作流其中节点Node就是你的Fleet成员边Edge定义了节点之间的流转条件。一个简单的顺序协作Fleet其图可能是一条直线开始 - Agent A - Agent B - Agent C - 结束。 一个复杂的决策型Fleet图可能包含分支和循环根据Agent A的输出结果决定是走向Agent B还是Agent C。LangSmith Fleet的价值在于它能深度集成并可视化这些由LangGraph创建的工作流将抽象的图转化为可监控、可调试的实时执行流水线。3.4 LangSmith集成舰队的指挥塔与黑匣子这是Fleet区别于你自己手写编排脚本的最大亮点。通过将你的FleetLangGraph工作流配置到LangSmith项目中你可以自动获得全链路追踪Trace每一次Fleet的执行都会生成一条完整的Trace。点击这条Trace你能逐层展开看到每个成员Agent的详细执行情况包括其子步骤。集中化日志与监控所有成员的调用日志、LLM的请求与响应、工具的输出都汇聚在一个地方。你可以设置警报监控延迟、错误率和成本。测试与评估你可以针对整个Fleet创建测试数据集运行批量测试评估其端到端的性能而不仅仅是单个组件。协作与共享团队可以共同查看和分析Fleet的运行情况加速调试和迭代过程。4. 实战构建你的第一个LangSmith Fleet理论说了这么多我们来动手搭建一个简单的Fleet。假设我们要构建一个“技术博客助手Fleet”它的任务是根据一个模糊的技术概念描述先搜索最新的相关信息然后生成一份结构清晰的报告大纲。这个Fleet将由两个成员组成Research Agent研究Agent负责使用搜索引擎工具获取最新信息。Outline Agent大纲Agent负责根据研究结果生成一份报告大纲。我们将使用LangGraph来编排它们。4.1 环境准备与依赖安装首先确保你的环境已就绪。# 安装核心库 pip install langchain langchain-community langgraph langsmith # 安装可能需要的工具包比如用于搜索的Tavily这里以Tavily为例你也可以用SerpAPI等 pip install tavily-python接下来你需要设置好LangSmith的API密钥。如果你还没有账号需要先去LangSmith官网注册。import os from getpass import getpass # 设置环境变量 os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_ENDPOINT] https://api.smith.langchain.com # LangSmith API端点 os.environ[LANGCHAIN_API_KEY] getpass(请输入你的LangSmith API Key: ) os.environ[LANGCHAIN_PROJECT] My-First-Fleet # 指定项目名在LangSmith UI中查看 # 设置Tavily搜索API Key (或其他工具Key) os.environ[TAVILY_API_KEY] getpass(请输入你的Tavily API Key: )4.2 定义舰队成员Agents我们先创建两个独立的Agent作为舰队成员。from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults # 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 1. 创建 Research Agent research_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的技术研究员。根据用户提供的技术概念或问题使用搜索工具查找最新的、权威的信息。请总结搜索到的关键点并注明信息来源。), (user, {input}) ]) research_tools [TavilySearchResults(max_results3)] # 限制3条结果避免信息过载 research_agent create_tool_calling_agent(llm, research_tools, research_prompt) research_agent_executor AgentExecutor(agentresearch_agent, toolsresearch_tools, verboseFalse) # 2. 创建 Outline Agent outline_prompt ChatPromptTemplate.from_messages([ (system, 你是一个技术文档架构师。根据研究员提供的信息摘要生成一份详细、结构清晰的技术报告或博客文章大纲。大纲应包含引言、核心章节分点论述、总结等部分。), (user, 请基于以下研究信息生成一份报告大纲\n\n{research_summary}) ]) # Outline Agent 不需要工具它只做文本生成 outline_agent outline_prompt | llm # 使用LCEL语法创建一个简单的Chain4.3 使用LangGraph构建舰队编排现在我们用LangGraph将这两个成员连接起来定义工作流。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator # 首先定义Fleet的共享状态State class FleetState(TypedDict): # 用户原始输入 input: str # Research Agent 的产出 research_summary: str # Outline Agent 的产出 final_outline: str # 初始化图 workflow StateGraph(FleetState) # 定义节点函数即我们的成员执行逻辑 def call_research_agent(state: FleetState): 节点1执行研究Agent print(f[Research Agent] 正在研究: {state[input]}) result research_agent_executor.invoke({input: state[input]}) return {research_summary: result[output]} def call_outline_agent(state: FleetState): 节点2执行大纲Agent print(f[Outline Agent] 正在基于研究生成大纲...) result outline_agent.invoke({research_summary: state[research_summary]}) return {final_outline: result.content} # 将节点添加到图中 workflow.add_node(research, call_research_agent) workflow.add_node(outline, call_outline_agent) # 设置边的流转逻辑research - outline - END workflow.set_entry_point(research) workflow.add_edge(research, outline) workflow.add_edge(outline, END) # 编译图得到我们的Fleet fleet_app workflow.compile()4.4 运行舰队并与LangSmith集成现在我们可以运行这个Fleet了。由于我们已经设置了LANGCHAIN_TRACING_V2环境变量这次调用会自动被记录到LangSmith的My-First-Fleet项目中。# 定义初始输入 initial_state {input: 解释一下LangGraph中的持久化检查点Checkpoint机制是如何工作的, research_summary: , final_outline: } # 运行Fleet final_state fleet_app.invoke(initial_state) print(\n *50) print(【任务完成】) print(f用户问题{final_state[input]}) print(f研究摘要\n{final_state[research_summary][:300]}...) # 截取部分显示 print(f生成的大纲\n{final_state[final_outline]})运行这段代码后立刻打开LangSmith网站进入你的My-First-Fleet项目。你应该能看到一条新的Trace记录。点击进去你会看到一幅清晰的图一个顶层的langgraph节点代表整个Fleet工作流。展开后能看到research和outline两个子节点。进一步展开research节点你能看到其中包含的LLM调用、工具TavilySearch调用等所有细节。同样outline节点的LLM调用详情也一览无余。这就是LangSmith Fleet提供的集中式可观测性。所有环节尽在掌握。5. 高级模式与最佳实践一个简单的顺序流只是开始。LangSmith Fleet的真正威力在于处理复杂逻辑。5.1 条件路由与循环假设我们的Fleet需要做决策如果研究Agent发现信息不足就触发第二轮更精确的搜索否则直接生成大纲。这需要条件路由。from langgraph.graph import StateGraph, END from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver # 假设我们定义了一个判断信息是否充足的函数 def is_info_sufficient(state: FleetState) - str: 根据研究摘要判断信息是否充足 summary state[research_summary] # 这里可以是一个简单的规则也可以调用一个LLM来判断 if len(summary) 100 or 信息不足 in summary: # 示例规则 return need_more_research else: return enough_for_outline # 创建新图 conditional_workflow StateGraph(FleetState) # 添加节点复用之前的函数 conditional_workflow.add_node(research, call_research_agent) conditional_workflow.add_node(outline, call_outline_agent) # 设置入口 conditional_workflow.set_entry_point(research) # 关键设置条件边。从research节点出来后根据函数决定下一步 conditional_workflow.add_conditional_edges( research, is_info_sufficient, # 路由判断函数 { need_more_research: research, # 跳回research节点形成循环 enough_for_outline: outline, } ) conditional_workflow.add_edge(outline, END) conditional_fleet conditional_workflow.compile()在这个流程中LangSmith的Trace会清晰地展示出执行路径你可以看到research节点是否被重复执行这对于调试循环逻辑至关重要。5.2 并行执行与聚合有些任务可以并行处理以提升效率。例如一个“竞品分析Fleet”可能需要同时搜索关于产品A、B、C的信息。from typing import List from langgraph.graph import StateGraph, END import asyncio class ParallelState(TypedDict): topics: List[str] research_results: List[str] combined_analysis: str def research_one_topic(topic: str) - str: 模拟针对一个主题的研究 # 这里可以封装一个专用的研究agent return f关于{topic}的研究摘要... def parallel_research_node(state: ParallelState): 并行研究节点 topics state[topics] # 使用线程池或asyncio进行并行调用示例为简单同步模拟 results [research_one_topic(topic) for topic in topics] return {research_results: results} def analysis_node(state: ParallelState): 分析聚合节点 all_results \n---\n.join(state[research_results]) # 调用LLM进行综合分析 analysis f综合报告\n{all_results} # 简化处理 return {combined_analysis: analysis} parallel_workflow StateGraph(ParallelState) parallel_workflow.add_node(parallel_research, parallel_research_node) parallel_workflow.add_node(analyze, analysis_node) parallel_workflow.set_entry_point(parallel_research) parallel_workflow.add_edge(parallel_research, analyze) parallel_workflow.add_edge(analyze, END) parallel_fleet parallel_workflow.compile()在LangSmith中parallel_research节点内部的并行调用可能会被展示为多个并发的子轨迹让你能同时监控多个任务的进度。5.3 错误处理与韧性一个健壮的Fleet必须能处理成员失败的情况。LangGraph允许你为节点设置try-catch式的错误处理。from langgraph.graph import StateGraph, END from langgraph.types import Command, interrupt class RobustState(TypedDict): input: str result: str error: str def unreliable_node(state: RobustState): 一个可能失败的节点 import random if random.random() 0.3: # 30%概率失败 raise ValueError(模拟随机失败) return {result: 任务成功完成} def fallback_node(state: RobustState): 降级处理节点 return {result: f主流程失败使用降级方案。原始输入{state[input]}} robust_workflow StateGraph(RobustState) # 添加主节点并指定其错误处理策略 robust_workflow.add_node( main_task, unreliable_node, # 使用 interrupt 和 Command 处理错误重试或转向fallback # 这里是一个简化示例实际使用可能需要更复杂的 try_except 装饰器或自定义逻辑 ) robust_workflow.add_node(fallback, fallback_node) # 一种常见模式在边逻辑中判断状态中是否有error字段决定走向 def route_after_main(state: RobustState): if state.get(error): return fallback return END robust_workflow.set_entry_point(main_task) # 注意实际错误处理配置更复杂这里仅为说明概念 # 可能需要使用 add_node 的 handlers 参数或自定义装饰器在LangSmith中你可以通过筛选“Error”状态的Trace快速定位和复盘Fleet中发生的故障查看具体的错误信息和堆栈加速排错过程。5.4 持久化与状态管理对于长时间运行或需要暂停/恢复的FleetLangGraph集成了**检查点Checkpoint**机制。这允许你将Fleet的完整状态包括每个节点的中间输出保存到数据库如SQLite、Postgres之后可以从中断点继续执行。from langgraph.checkpoint.sqlite import SqliteSaver # 创建支持检查点的图 memory SqliteSaver.from_conn_string(:memory:) # 使用内存数据库生产环境需用文件或网络DB persistent_workflow StateGraph(FleetState) persistent_workflow.add_node(research, call_research_agent) persistent_workflow.add_node(outline, call_outline_agent) persistent_workflow.set_entry_point(research) persistent_workflow.add_edge(research, outline) persistent_workflow.add_edge(outline, END) # 编译时传入检查点存储器 persistent_fleet persistent_workflow.compile(checkpointermemory) # 调用时会生成一个唯一的线程IDthread_id config {configurable: {thread_id: user_123_session_1}} try: # 第一次执行可能在research后中断 state persistent_fleet.invoke(initial_state, config) except Exception as e: print(f执行中断: {e}) # 状态已被自动保存到检查点 # 稍后根据thread_id恢复执行 later_state persistent_fleet.invoke({input: 继续}, config) # 输入可能被忽略因为状态已保存这对于处理需要用户多次交互如多轮对话的复杂Agent Fleet来说是必不可少的功能。LangSmith同样可以追踪带有检查点的执行让你看到跨会话的完整历史。6. 在LangSmith UI中管理与优化你的Fleet构建Fleet只是第一步持续的监控、调试和优化才是发挥其价值的关键。LangSmith UI为此提供了强大支持。6.1 可视化追踪与调试Trace详情视图这是你调试的主要战场。对于每一次Fleet调用你可以查看执行图直观地看到工作流中每个节点的执行顺序、状态成功/失败和耗时。钻取节点细节点击任何一个节点如research查看其详细的输入输出、内部的LLM调用、工具调用记录、Token消耗和成本。对比运行结果如果你修改了Prompt或模型可以并排对比不同版本Fleet在同一条输入下的执行轨迹和结果精准定位变化。搜索与筛选你可以根据时间、状态成功/错误、输入关键词、甚至某个特定节点的输出来筛选Trace快速找到问题所在。例如搜索“error”来查找所有失败的运行。6.2 测试与评估“langsmith生成报告”这个热词很可能指向了LangSmith的**数据集Dataset和测试Testing**功能。这对于Fleet的迭代至关重要。创建数据集将一系列典型的用户查询如“帮我分析一下LangSmith Fleet的优缺点”、“写一个关于RAG的教程大纲”整理成数据集。运行批量测试在LangSmith UI中选择你的Fleet或对应的LangGraph工作流和这个数据集启动一次批量测试。系统会自动用所有用例调用你的Fleet。生成评估报告测试完成后LangSmith可以自动或半自动地通过LLM作为评判员为每个结果打分生成一份综合报告。报告会展示Fleet的整体成功率、平均延迟、成本并高亮出哪些用例失败了帮助你系统性评估Fleet的质量和稳定性。6.3 监控与告警在项目设置中你可以为你的Fleet项目配置监控看板和告警。看板查看Fleet的每日调用量、平均延迟、错误率、总成本等关键指标的趋势图。告警设置规则例如“当错误率在5分钟内超过5%时”或“当平均延迟超过10秒时”通过邮件、Slack等方式通知你。这让你能在用户大规模投诉前主动发现Fleet的性能退化或故障。6.4 协作与知识共享你可以将Fleet所在的LangSmith项目分享给团队成员。大家可以在同一条Trace下添加评论讨论某个奇怪输出产生的原因或者共同审查新版本的测试结果。这改变了AI应用开发“黑盒”和孤岛式的协作模式。7. 避坑指南与实战心得在将多个Agent编排成Fleet的实践中我踩过不少坑也总结出一些让Fleet更可靠、更高效的经验。7.1 状态设计是重中之重Fleet的共享状态State是成员间通信的唯一桥梁。设计一个糟糕的状态结构是灾难的开始。保持状态扁平与明确避免嵌套过深的字典。为每个信息片段起一个清晰的名字如user_query,search_results,draft_answer,final_output。区分“只读”与“可写”在团队协作中最好通过文档约定哪些节点会修改状态的哪个部分防止意外覆盖。虽然LangGraph不强制但清晰的约定能减少Bug。序列化兼容性如果你使用持久化检查点确保状态中的所有对象都是可序列化的如使用Pydantic的BaseModel。自定义类对象可能会导致保存和恢复失败。7.2 控制流复杂度与可调试性的权衡LangGraph非常强大可以构建带有复杂条件分支、循环和并行节点的图。但请记住越复杂的图越难调试。在添加一个新的条件分支或循环之前问自己是否真的必要。很多时候拆分成两个更简单、顺序执行的Fleet可能是更好的选择。为节点起有意义的名称。不要用node_1,node_2而用validate_input,search_web,synthesize_answer。这在查看LangSmith的Trace时能让你一眼看懂执行流程。善用“人工干预”节点对于关键决策点例如判断用户情绪非常负面可以设计一个节点其输出不是由LLM决定而是挂起流程等待人工审核输入。这能极大地增加高风险Fleet的可靠性。7.3 错误处理不要“一刀切”不是所有错误都需要整个Fleet崩溃。分级处理网络超时可以重试工具返回“未找到结果”可以触发备用搜索或进入降级逻辑LLM输出格式错误可以尝试修复或请求重生成。只有遇到无法恢复的错误如权限错误、关键逻辑异常才让整个Fleet失败。在状态中传递错误信息即使一个节点失败了也可以将错误信息和上下文写入状态让后续的fallback节点或“通知人工”节点能够获取到做出更合理的响应。7.4 性能与成本监控多Agent系统很容易在你不注意的时候产生高昂的成本和延迟。为每个LLM调用设置合理的max_tokens和temperature。在Fleet中一个节点的高消耗会被放大。在LangSmith中密切关注成本报告。定期查看哪个Agent、哪个LLM调用最“烧钱”并思考能否优化Prompt或使用更便宜的模型。考虑缓存对于内容变化不频繁的查询如“什么是机器学习”可以在Fleet入口处引入一个缓存层避免重复执行昂贵的搜索和LLM调用。LangChain也提供了多种缓存集成。7.5 从简单开始逐步迭代不要试图一开始就设计一个包含10个Agent、有5层条件判断的“终极Fleet”。先构建一个最小可行产品MVPFleet哪怕它只有两个节点顺序执行。在LangSmith上用它处理一批真实用例观察Trace找到瓶颈和失败点。基于数据和观察而不是猜想来增加复杂性。例如如果你发现research节点经常返回无关信息那就先优化这个节点的Prompt或工具而不是急于增加一个“信息过滤”节点。每次只做一个改动并用LangSmith的测试套件来评估这个改动是带来了提升还是引入了回归。LangSmith Fleet标志着一个新的阶段AI应用开发从 crafting prompts雕琢提示词和 wiring up chains连接链进入到了 orchestrating multi-agent systems编排多智能体系统的时代。它带来的最大改变是将复杂AI应用的构建从“艺术”和“玄学”更多地推向“工程”和“可观测性”。你不再是在黑暗中摸索单个组件的表现而是拥有了一个全景仪表盘能看清整个智能体团队是如何协作、在哪里卡壳、为何出错。对我而言最深刻的体会是它迫使开发者以更模块化、更工程化的思维来设计AI应用。每个Agent变得像微服务一样职责清晰接口明确。而LangSmith提供的追踪、测试和监控则是确保这个“微服务集群”能稳定运行的运维平台。如果你正在构建涉及多个步骤、需要不同专业能力的AI应用那么投入时间学习LangGraph和LangSmith Fleet将会在未来为你节省大量的调试时间和运维成本。