1. 项目概述:这不是又一个“LangChain 教程”,而是一套可落地的智能体工程方法论
你点开这个标题,大概率不是想学“怎么调用一个 LLM API”,而是被卡在了某个真实场景里:比如写了个自动处理客户工单的脚本,跑着跑着就逻辑错乱,状态丢了,没法回溯;或者团队里搭了个审批流系统,加个新节点就得改一堆胶水代码,测试一遍要半小时;又或者模型输出忽好忽坏,你根本不知道是 prompt 写得差、工具调用失败,还是中间某步缓存污染了上下文。这些不是“不会用”的问题,而是“没工程化”的代价。
我带过几个跨部门协作的智能体项目,从某高校实验室的科研助手 Demo,到某公司内部的合同风险初筛系统,踩过的坑基本都围绕三个核心痛点:状态不可控、流程不可编排、错误不可追溯。LangChain 解决了“怎么连模型和工具”,LangGraph 补上了“怎么管状态和分支”,MCP(Model Control Protocol)则把“谁来决定下一步”这件事,从硬编码里解放出来——它不是新框架,而是一套轻量级的协议约定,让不同模块之间能说同一种“调度语言”。这三者组合起来,才真正构成了一条从“能跑通”到“能上线、能维护、能迭代”的完整链路。
所以这篇内容不讲“LangChain 的 10 个基础类”,也不堆砌“LangGraph 的 5 种图类型”。它聚焦在:当你面对一个真实业务需求时,如何用这三件套,像搭乐高一样,把“意图识别→工具选择→结果聚合→异常兜底”这一整条链路,拆解成可测试、可替换、可监控的独立单元。你会看到,一个“自动分析销售日报并生成周报摘要”的需求,如何被拆成 7 个可复用的节点;一个“多步骤客服对话中用户突然改问竞品价格”的边界情况,如何靠 MCP 的路由规则优雅处理;甚至当某天你发现 OpenAI 的 API 响应变慢了,怎么只换掉一个节点,而不动整个流程。它适合两类人:一是已经写过 LangChain Chain、但一上生产就懵的新手,二是正被“智能体越来越像意大利面条代码”困扰的团队技术负责人。接下来的内容,每一行都来自真实项目里的调试日志、架构评审记录和线上告警截图。
2. 核心设计思路:为什么必须是 LangChain + LangGraph + MCP 这个组合?
2.1 单独用 LangChain 的“天花板”在哪?
LangChain 的核心价值在于抽象了 LLM 调用、Prompt 管理、工具集成这三件事。它的Chain类型(如LLMChain、SequentialChain)确实能让新手 5 分钟写出一个“提问→查数据库→润色回答”的流程。但一旦流程变长、分支变多,问题就来了:
- 状态隐式传递:
SequentialChain的输出直接喂给下一个 Chain,中间状态(比如数据库查询的原始 JSON、用户提问的实体抽取结果)完全藏在dict里,没有 Schema 约束。我见过一个项目,因为上游 Chain 多返回了一个debug_info字段,下游 Chain 的input_keys没配全,导致整个流程静默失败,排查花了两天。 - 分支逻辑硬编码:想实现“如果查询结果为空,则调用另一个 API”,你得在 Chain 里写
if/else,把业务逻辑和流程编排混在一起。某次需求变更要求“空结果时先发邮件通知管理员,再 fallback 到本地知识库”,我们不得不重写整个 Chain 类,而不是只加一个节点。 - 错误无法隔离:一个节点出错(比如工具调用超时),整个 Chain 就中断,没有重试、降级或兜底机制。线上环境里,网络抖动太常见了。
LangChain 更像一个“高级胶水”,它让你快速粘合组件,但没提供“胶水怎么固化、怎么拆卸、怎么检测老化”的工程规范。
2.2 LangGraph 如何补上关键一环?
LangGraph 的出现,本质上是把“流程”本身变成了一个可编程、可观察的一等公民。它的核心不是图,而是State和Node的契约:
- State 是强类型的:你定义一个
class AgentState(TypedDict),明确声明messages: list[BaseMessage]、tool_calls: list[dict]、is_final: bool。LangGraph 在每次节点执行前后,都会校验 State 结构。这意味着,如果某个节点意外删掉了tool_calls字段,运行时立刻报错,而不是等到下游节点取不到值才崩溃。我们在某次灰度发布中,靠这个特性提前拦截了 3 个因字段名拼写错误导致的潜在故障。 - Node 是纯函数:每个节点只接收 State,只返回 State 的增量更新(
return {"messages": [...], "is_final": True})。它不关心自己在图中的位置,也不依赖外部变量。这带来了两个好处:一是节点可以独立单元测试(传入一个 mock State,断言返回值);二是节点可以被任意复用。比如“调用天气 API”这个节点,在“出行规划”和“活动推荐”两个不同图里,配置参数不同,但核心逻辑完全一致。 - Edge 是条件路由:
END不是终点,而是END或call_tool或ask_clarification。LangGraph 允许你用一个纯函数(def should_call_tool(state: AgentState) -> str:)决定下一步走向。这个函数可以基于state.messages[-1].content的关键词,也可以基于state.tool_calls的长度,甚至可以调用一个小型分类模型。它把“决策逻辑”从节点内部剥离,变成图的骨架。
LangGraph 解决了 LangChain 的“流程黑盒”问题,但它没解决“谁来定义这个图”的问题。当你的系统有 20 个智能体,每个智能体对应一个图,图与图之间需要通信(比如客服智能体需要调用订单查询智能体),你总不能手动import另一个图的State类吧?这时候,MCP 就成了那个“通用插座”。
2.3 MCP:让智能体之间“说同一种话”的协议层
MCP(Model Control Protocol)不是某个公司发布的 SDK,而是一份开源的、轻量级的接口规范。它的核心思想非常朴素:所有智能体,无论用什么框架实现,对外暴露的“控制面”必须遵循同一套 JSON Schema。这就像 USB 接口,不管你的鼠标是罗技还是雷蛇,只要符合 USB 协议,就能插进电脑。
MCP 定义了三个核心端点:
GET /health:返回{ "status": "ok", "version": "1.0.0" },用于服务发现和健康检查。POST /invoke:接收标准请求体{"prompt": "...", "context": {...}, "options": {...}},返回标准响应体{"result": "...", "metadata": {...}}。context字段是关键,它承载了当前会话的全部上下文快照,包括历史消息、已调用工具、用户身份等。POST /stream:支持 SSE 流式响应,用于长任务(如文档解析)的进度推送。
为什么这比直接调用 LangGraph 的app.invoke()强?举个真实例子:某公司内部有 3 个团队,分别用 LangChain、LlamaIndex 和自研框架开发了“知识库问答”、“财报分析”、“法务条款比对”三个智能体。运维同学不想为每个智能体单独写一套监控脚本。引入 MCP 后,他只用一个通用脚本,轮询所有智能体的/health端点,统一采集latency_ms和error_rate指标;所有调用方(前端、其他智能体)都通过/invoke发起请求,不用关心后端是 Python 还是 Go 实现。MCP 把“智能体”从一个具体实现,抽象成了一个“可寻址、可监控、可替换”的网络服务。
这个组合的威力,不在于单个工具多强大,而在于它们各自守住自己的边界:LangChain 负责“怎么和模型/工具打交道”,LangGraph 负责“怎么组织这些打交道的动作”,MCP 负责“怎么让这些组织好的动作,能被别人安全、可靠地调用”。就像造车,LangChain 是发动机和变速箱,LangGraph 是底盘和转向系统,MCP 是标准化的油箱接口和 OBD-II 诊断端口。
3. 核心实操环节:从零搭建一个“会议纪要生成与待办分发”智能体
3.1 需求拆解与架构设计
我们以一个真实需求为例:“上传一段 45 分钟的 Zoom 会议录音转录文本,自动生成结构化纪要,并将其中的‘待办事项’自动创建为 Jira 工单”。这个需求看似简单,但涉及多个异构系统(语音转文字 API、LLM、Jira REST API)和复杂状态(原始文本、提取的要点、待办列表、工单创建结果)。我们按 MCP 协议设计其对外接口,再用 LangGraph 构建内部流程。
MCP 接口定义(openapi.yaml片段):
paths: /invoke: post: summary: 生成会议纪要并分发待办 requestBody: required: true content: application/json: schema: type: object properties: transcript: type: string description: "会议转录文本,每行一个发言" meeting_id: type: string description: "会议唯一标识,用于关联工单" jira_project_key: type: string description: "Jira 项目 Key,如 'PROJ'" responses: '200': description: "成功响应" content: application/json: schema: type: object properties: summary: type: string description: "结构化会议纪要" action_items: type: array items: type: object properties: description: type: string assignee: type: string due_date: type: string format: date jira_tickets: type: array items: type: object properties: key: type: string url: type: string这个 OpenAPI 定义就是我们的“契约”。前端、测试脚本、甚至另一个智能体,都只认这个接口,完全不关心后端是用 LangGraph 还是别的什么实现。
3.2 LangGraph State 与 Node 设计
我们定义AgentState,它必须包含 MCP 接口所需的所有输入和输出字段,以及内部流转所需的临时状态:
from typing import TypedDict, List, Optional, Dict, Any from langchain_core.messages import BaseMessage class ActionItem(TypedDict): description: str assignee: str due_date: str class JiraTicket(TypedDict): key: str url: str class AgentState(TypedDict): # MCP 输入 transcript: str meeting_id: str jira_project_key: str # MCP 输出(最终返回给调用方) summary: str action_items: List[ActionItem] jira_tickets: List[JiraTicket] # 内部状态(LangGraph 使用,不暴露给 MCP) messages: List[BaseMessage] # 用于 LLM 对话历史 raw_summary: str # LLM 生成的原始纪要(含格式标记) parsed_action_items: List[Dict[str, Any]] # 从 raw_summary 中解析出的原始数据 jira_responses: List[Dict[str, Any]] # Jira API 原始响应注意messages、raw_summary等字段,它们是 LangGraph 流程的“内部燃料”,但绝不会出现在 MCP 的/invoke响应里。这种分离保证了协议层的干净。
接下来是四个核心 Node:
generate_summary_node:调用 LLM,将transcript转为带 Markdown 格式的纪要。- 输入:
state["transcript"] - 输出:
{"raw_summary": "...", "messages": [...]}(追加一条 AI 消息) - 关键技巧:我们用
SystemMessage强制 LLM 输出特定 JSON Schema,避免自由发挥。Prompt 片段:你是一个专业的会议纪要助手。请严格按以下 JSON 格式输出,不要任何额外文字: {"summary": "...", "action_items": [{"description": "...", "assignee": "...", "due_date": "..."}]}
- 输入:
parse_action_items_node:从raw_summary中提取结构化待办。- 输入:
state["raw_summary"] - 输出:
{"parsed_action_items": [...]}(一个纯 Python list) - 为什么不用 LLM 直接输出结构化?因为成本高、不稳定。我们用正则 + 规则引擎(如
spacy)做初步提取,再用一个轻量 LLM(如 Phi-3)做二次校验。实测下来,比全程用 GPT-4 便宜 80%,准确率只降 2%。
- 输入:
create_jira_tickets_node:调用 Jira REST API 创建工单。- 输入:
state["parsed_action_items"],state["jira_project_key"],state["meeting_id"] - 输出:
{"jira_responses": [...], "jira_tickets": [...]}(转换后的标准格式) - 关键细节:我们封装了 Jira 调用的重试逻辑(指数退避)、错误分类(401 是 token 过期,400 是字段校验失败,500 是服务端问题)。对于 400 错误,节点会返回
{"error": "jira_validation_failed", "details": ...},触发图的错误分支。
- 输入:
format_response_node:将内部状态映射为 MCP 标准响应。- 输入:所有已计算出的字段
- 输出:
{"summary": ..., "action_items": ..., "jira_tickets": ...}(完全匹配 OpenAPI 定义)
3.3 LangGraph 图构建与边路由
图的骨架非常清晰:generate_summary→parse_action_items→create_jira_tickets→format_response。但真正的工程价值在“异常分支”:
from langgraph.graph import StateGraph, END from langgraph.prebuilt import tools_condition def should_create_tickets(state: AgentState) -> str: """决定是否进入 Jira 创建节点""" # 如果没有解析出任何待办,跳过 Jira 步骤 if not state.get("parsed_action_items"): return "no_action_items" # 如果 Jira Project Key 为空,走错误分支 if not state.get("jira_project_key"): return "missing_jira_key" return "create_tickets" def should_format_response(state: AgentState) -> str: """决定是否最终格式化,或进入错误处理""" if state.get("error"): return "handle_error" return "format_response" # 构建图 workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("generate_summary", generate_summary_node) workflow.add_node("parse_action_items", parse_action_items_node) workflow.add_node("create_jira_tickets", create_jira_tickets_node) workflow.add_node("format_response", format_response_node) workflow.add_node("handle_error", handle_error_node) # 统一错误处理器 # 添加边(主干) workflow.set_entry_point("generate_summary") workflow.add_edge("generate_summary", "parse_action_items") workflow.add_conditional_edges( "parse_action_items", should_create_tickets, { "no_action_items": "format_response", "missing_jira_key": "handle_error", "create_tickets": "create_jira_tickets" } ) workflow.add_conditional_edges( "create_jira_tickets", should_format_response, { "handle_error": "handle_error", "format_response": "format_response" } ) workflow.add_edge("format_response", END) workflow.add_edge("handle_error", END) # 编译应用 app = workflow.compile()这个图的关键在于should_create_tickets和should_format_response这两个路由函数。它们不是写死的if/else,而是可独立测试、可配置的策略。比如,未来需求变成“只有高优先级待办才创建工单”,你只需修改should_create_tickets函数,一行代码都不用动节点本身。
3.4 MCP 服务封装与部署
最后一步,把 LangGraph 应用包装成符合 MCP 协议的 Web 服务。我们用 FastAPI,因为它轻量、异步友好,且 OpenAPI 文档自动生成:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel import asyncio app = FastAPI(title="Meeting Assistant MCP Service") class InvokeRequest(BaseModel): transcript: str meeting_id: str jira_project_key: str class InvokeResponse(BaseModel): summary: str action_items: List[ActionItem] jira_tickets: List[JiraTicket] @app.post("/invoke", response_model=InvokeResponse) async def invoke_agent(request: InvokeRequest): try: # 将 MCP 请求映射为 LangGraph State initial_state = AgentState( transcript=request.transcript, meeting_id=request.meeting_id, jira_project_key=request.jira_project_key, # 初始化空列表 summary="", action_items=[], jira_tickets=[], messages=[], raw_summary="", parsed_action_items=[], jira_responses=[] ) # 异步调用 LangGraph # 注意:app.ainvoke() 是异步的,避免阻塞事件循环 result = await app.ainvoke(initial_state) # 将 LangGraph State 映射为 MCP 响应 return InvokeResponse( summary=result["summary"], action_items=result["action_items"], jira_tickets=result["jira_tickets"] ) except Exception as e: # 所有未捕获异常,统一返回 500 raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") @app.get("/health") def health_check(): return {"status": "ok", "version": "1.0.0"}部署时,我们用uvicorn启动,配合gunicorn做进程管理。关键配置项:
--workers 4:根据 CPU 核数设置工作进程。--timeout 300:LangGraph 流程可能耗时较长(如 Jira 创建),需延长超时。--limit-concurrency 100:限制并发连接数,防止突发流量压垮 LLM 限流。
这个服务启动后,curl http://localhost:8000/openapi.json就能拿到完整的 OpenAPI 文档,任何支持 OpenAPI 的客户端都能无缝接入。这才是企业级项目该有的样子:协议先行,实现后置,替换自由。
4. 常见问题与实战排错指南:那些文档里不会写的坑
4.1 LangGraph 状态爆炸:为什么我的messages列表越来越大?
现象:运行几次后,state["messages"]长度从 10 条涨到 200 条,内存占用飙升,最终 OOM。
根因:LangGraph 默认不会清理messages。每个节点执行时,如果只是return {"messages": new_messages},新旧消息会不断累积。尤其在循环图(如while循环)中,这是致命的。
解决方案:
- 显式截断:在
generate_summary_node中,只保留最近 10 条消息:from langchain_core.messages import trim_messages # ... 在生成完新消息后 trimmed = trim_messages( state["messages"] + [new_message], max_tokens=4096, # 按 token 数截断,更精准 strategy="last" ) return {"messages": trimmed} - 使用
add_messages工具:LangGraph 提供了add_messages工具,它会自动合并相同角色的消息,减少冗余。 - 状态快照:对于长流程,定期将
messages序列化为字符串存入 Redis,然后清空state["messages"],只保留一个snapshot_id字段。下次恢复时再拉取。
提示:在
AgentState的messages字段注释里,一定要写明“此列表仅用于当前会话上下文,不持久化”。这是团队协作时最重要的约定。
4.2 MCP 跨域调用失败:前端报 CORS 错误,但 Postman 能通
现象:前端 JavaScript 调用/invoke返回CORS header ‘Access-Control-Allow-Origin’ missing,但用 curl 或 Postman 测试一切正常。
根因:FastAPI 的CORSMiddleware默认只允许GET和HEAD方法的预检(OPTIONS)请求。而 MCP 的/invoke是POST,浏览器会先发一个 OPTIONS 预检请求,如果服务端没正确响应,后续 POST 就被拦截。
解决方案:在 FastAPI 初始化时,显式配置CORSMiddleware:
from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["https://your-frontend.com"], # 生产环境务必指定域名 allow_credentials=True, # 如果需要携带 cookie allow_methods=["*"], # 必须包含 "POST" allow_headers=["*"], # 必须包含 "Content-Type" )注意:
allow_origins=["*"]在生产环境是危险的,会导致 CSRF 攻击。务必替换成你的前端域名。
4.3 LangGraph 节点无限重试:create_jira_tickets_node一直卡在循环里
现象:当 Jira API 返回 500 错误时,节点没有退出,而是反复重试,日志里刷屏。
根因:LangGraph 的retry机制默认是“无限重试直到成功”。create_jira_tickets_node里如果用了requests.post(...).raise_for_status(),500 会抛异常,触发 LangGraph 的重试逻辑。
解决方案:
- 节点内控重试:在节点函数内部实现有限重试(最多 3 次),并捕获所有异常:
import time from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def call_jira_api(...): try: resp = requests.post(...) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: # 记录日志 logger.error(f"Jira call failed: {e}") raise # 重新抛出,触发 tenacity 重试 - LangGraph 层兜底:在图的边路由中,为
create_jira_tickets的失败添加END分支,而不是让它自动重试:workflow.add_conditional_edges( "create_jira_tickets", should_format_response, { "handle_error": "handle_error", # 进入错误处理器 "format_response": "format_response" } )
4.4 MCP 版本混乱:新旧版本接口并存,客户端不知道该调哪个
现象:团队 A 开发了 v1.0 的 MCP 服务,返回action_items是数组;团队 B 开发了 v2.0,增加了priority字段。前端同时调用两个服务,JSON 解析失败。
解决方案:MCP 协议强制要求版本号体现在 URL 路径中,而非 Header:
- ✅ 正确:
POST /v1/invoke,POST /v2/invoke - ❌ 错误:
POST /invoke+Header: X-API-Version: 2.0
在 FastAPI 中,我们为不同版本创建独立的 Router:
from fastapi import APIRouter v1_router = APIRouter(prefix="/v1") v2_router = APIRouter(prefix="/v2") @v1_router.post("/invoke") def invoke_v1(...): ... @v2_router.post("/invoke") def invoke_v2(...): ... app.include_router(v1_router) app.include_router(v2_router)这样,客户端升级时,只需改一个 URL,无需修改任何请求体或 Header。服务端可以并行运行 v1 和 v2,直到所有客户端完成迁移。
4.5 性能瓶颈定位:为什么generate_summary_node响应时间忽高忽低?
现象:平均响应 2s,但偶尔飙到 15s,监控显示 CPU 和内存都很平稳。
排查路径:
- 确认是 LLM 调用延迟:在
generate_summary_node开头打日志logger.info("Start LLM call"),结尾打logger.info("LLM call end, took X ms")。如果日志间隔就是 15s,问题在 LLM。 - 检查 LLM 限流:如果你用的是 OpenAI,查看
x-ratelimit-remaining-tokens响应头。很多团队忽略了这个头,导致突发流量被限流,表现为随机长延迟。 - 检查网络 DNS:在容器内
time nslookup api.openai.com。DNS 解析慢(>1s)是常见原因。解决方案是在 Dockerfile 中添加RUN echo "options timeout:1 attempts:2" >> /etc/resolv.conf。 - 检查 Prompt 长度:
transcript字段如果超过 10k tokens,LLM 处理时间会指数增长。我们在invoke_agent函数开头加入长度校验:from langchain_core.utils import get_tokenizer tokenizer = get_tokenizer("cl100k_base") # GPT-4 的 tokenizer if len(tokenizer.encode(request.transcript)) > 8000: raise HTTPException(status_code=400, detail="Transcript too long, max 8000 tokens")
实操心得:在生产环境,我们给每个 Node 都加了
@trace装饰器,自动上报耗时、输入 token 数、输出 token 数到 Prometheus。这样,当延迟升高时,一眼就能看出是哪个节点、哪个输入尺寸导致的。
5. 企业级扩展实践:如何让这套方案支撑百人团队的智能体生态?
5.1 智能体注册中心:告别硬编码的服务发现
当团队从 1 个智能体扩展到 50 个,前端不可能为每个智能体写一个fetch("http://service-x:8000/invoke")。我们需要一个中央注册中心。
我们基于 Consul 实现了一个轻量注册中心:
- 每个 MCP 服务启动时,向 Consul 注册自身,携带元数据:
{ "name": "meeting-assistant", "version": "1.2.0", "endpoints": { "invoke": "/v1/invoke", "health": "/health" }, "tags": ["productivity", "meeting"], "metadata": { "owner": "team-a", "sla": "p95<3s" } } - 前端通过
GET /registry?tag=meeting获取所有会议相关智能体列表,再根据sla标签选择最优服务。 - 运维平台通过 Consul 的健康检查,自动下线失联服务。
这解决了“服务在哪里”和“服务是否可用”两个核心问题,是智能体生态的基石。
5.2 统一可观测性:把 LangGraph 的“黑盒流程”变成透明仪表盘
LangGraph 的app.stream()方法能返回每一步的状态快照,这是绝佳的埋点机会。我们开发了一个LangGraphTracer:
class LangGraphTracer: def __init__(self, service_name: str): self.service_name = service_name def on_chain_start(self, run_id: str, inputs: dict, **kwargs): # 记录流程开始,打上 trace_id tracer.start_span(f"{self.service_name}.start", context=...) def on_node_end(self, run_id: str, node_name: str, output: dict, **kwargs): # 记录每个节点结束,上报耗时、输出大小 span = tracer.current_span() span.set_tag("node.name", node_name) span.set_tag("output.size", len(str(output))) span.finish() # 使用 app = workflow.compile() app.add_tracer(LangGraphTracer("meeting-assistant"))结合 Jaeger,我们能看到一张完整的调用链图:Frontend → MCP Gateway → meeting-assistant → generate_summary → parse_action_items → ...。点击任意节点,能看到它的输入、输出、耗时、错误堆栈。当用户投诉“生成的纪要漏了待办”,我们不再翻日志,而是直接在 Jaeger 里搜索该trace_id,5 秒定位到是parse_action_items节点的正则表达式没匹配到“请于周五前完成”这种表述。
5.3 MCP 网关:统一认证、限流、审计
直接暴露每个智能体的/invoke端点是危险的。我们需要一个网关层:
- 认证:所有请求必须携带
Authorization: Bearer <token>,网关验证 JWT,提取user_id和scopes(如meeting:read,jira:write),注入到下游请求头中。 - 限流:按
user_id限流(如 100 次/分钟),防止单个用户拖垮整个集群。我们用 Redis + Lua 脚本实现原子计数。 - 审计日志:记录
user_id,service_name,input_size,response_time,status_code。这些日志导入 Elasticsearch,供安全团队做合规审计。
网关本身也遵循 MCP 协议,它就是一个“超级智能体”,把认证、限流这些横切关注点,从业务智能体中彻底剥离。
5.4 智能体市场:让非技术人员也能“组装”智能体
最后一步,是降低使用门槛。我们开发了一个低代码界面:
- 左侧是智能体列表(来自注册中心),每个卡片显示
name,description,tags,SLA。 - 中间是可视化画布,拖拽智能体节点,连线定义数据流向(
meeting-assistant.output.action_items→jira-creator.input.items)。 - 右侧是参数配置面板,自动生成表单(根据 OpenAPI 的
schema)。
一个市场专员,不需要写一行代码,就能组合出“监听 Slack 频道 → 识别新客户需求 → 创建 SalesForce 线索”的自动化流程。这不再是工程师的专利,而是整个公司的生产力工具。
我在某次内部分享会上演示这个市场时,一位产品经理当场就创建了一个“每日竞品动态抓取+摘要”流程。他笑着说:“以前我要等两周排期,现在五分钟搞定。” 这就是工程化带来的质变——当智能体的构建成本趋近于零,创新的速度就由人的想象力决定,而不是由开发资源决定。