用 AI Agent 构建自动化工作流：从原型到落地

2026/10/54 分钟阅读

用 AI Agent 构建自动化工作流

构建一套能稳定跑起来的 AI Agent 自动化工作流，核心不在于模型本身有多聪明，而在于如何定义工具（Tools）以及如何管理上下文与状态。AI Agent 的本质是「智能路由器」——它接收目标，根据当前状态选择调用哪个 API 或执行哪段代码，直到闭环达成。

为什么传统的脚本化自动化会失效？

在我之前的工作流设计中，我们习惯用 If-Then 规则引擎来编排任务。但当面对一个复杂的「数据汇总」需求时，传统方案就捉襟见肘了：比如需要去 CRM 拉取销售线索，还要去数据库查询该客户的历史订单，最后生成 PDF 报告并回写系统。

用 AI Agent 构建自动化工作流 - 插图 1

问题出在两个地方：

非结构化输入难以解析：CRM 导出的 Excel 列名经常变化，硬编码的正则表达式极易失效。
异常处理成本高：一旦某一步网络超时，整个任务直接失败，而实际上只需要换个 API 重试即可。

引入 AI Agent 后，我将每个 API 调用封装为独立 Tool，让 LLM 在运行时根据返回结果动态决定下一步。比如当 CRM 接口返回 429 限流错误时，Agent 能自主决定「稍后重试」而不是直接崩溃。

工作流架构：LCEL 与 MCP 的选型

在技术选型上，LangGraph 是目前实现多步 Agent 编排的首选框架。它通过「节点（Node）」和「边（Edge）」的状态机模式，让我能精确控制每个环节的流转逻辑。

我通常会按照以下三层结构设计 Agent 工作流：

感知层：使用 LLM 进行意图识别，将用户的模糊指令拆解为具体的子任务参数。这一步非常关键，它能解决「用户说得太简略」的问题。
决策层：通过 ReAct（推理-行动）循环驱动。每一轮迭代中，Agent 先思考（Reason）当前缺失什么信息，再行动（Act）调用工具获取。
执行层：实际的工具调用。这里强烈建议接入 MCP (Model Context Protocol) 标准。通过 MCP，我可以将数据库查询、搜索接口、甚至是内网服务统一封装为标准的 MCP 服务器，Agent 端只需声明依赖即可调用，彻底解决了工具集成的碎片化问题。

例如，我在给公司设计一个「月度销售报表自动生成」的 Agent 时，就是通过 MCP 连接了 PostgreSQL 和内部 BI 系统，Agent 能够自主决定何时去查明细数据、何时去查汇总数据，无需我预先写死查询顺序。

实战：处理 LLM 幻觉与状态回滚

构建 Agent 自动化流程时，最头疼的就是「假阳性」——Agent 自信地调用了不存在的函数，或者在中间步骤搞错了参数。

如何解决？

强制类型校验：不要在 LLM 侧做宽松校验。使用 Pydantic 或 TypeScript 对 Tool 的输入参数做严格的 Schema 定义。如果 LLM 传入了错误的参数类型，底层框架（如 LangGraph）会在执行前直接报错，而不是让它运行一半再抛出异常。
设置最大循环次数（Max Iterations）：默认值通常设为 10-20 次。一旦超过阈值强制停止，防止 Agent 陷入死循环或重复调用。
引入「人工审核节点」：对于涉及资金、关键数据写入等敏感操作，我在流程图里专门加了一个 Human-in-the-loop 节点。Agent 生成操作计划后，必须经过人工点击「Approved」，工作流才会继续。

以下是一个基于 LangGraph 的核心节点定义示例，展示了如何进行严格的参数校验与循环控制：

from typing import TypedDict
from langgraph.graph import StateGraph, END
from pydantic import BaseModel, Field

# 定义严格的状态 schema
class AgentState(TypedDict):
    iterations: int
    max_iterations: int
    user_input: str
    result: str

# 定义 Tool 参数 schema，防止 LLM 传错类型
class SearchInput(BaseModel):
    query: str = Field(..., description="搜索关键词")
    limit: int = Field(default=5, ge=1, le=20)

# 工具函数
async def search_tool(query: str, limit: int) -> str:
    # 实际业务逻辑
    return f"搜索结果: {query[:limit]}..."

# 构建图
builder = StateGraph(AgentState)

# 添加节点...
# 配置 max_iterations=10 防止无限循环
graph = builder.compile()

性能优化与成本控制

很多团队把 Agent 落地后，发现 API 调用费用远超预期。这主要是因为 Agent 的思考过程会消耗大量的 Token。

我的优化经验如下：

Context Pruning（上下文裁剪）：不要把所有历史对话都喂给模型。只保留最近 N 轮的对话，或者是工具调用的摘要。对于超长的工作流，考虑在每一步结束后把中间结果序列化为变量，逐步推进，而不是堆积上下文。
小模型做路由，大模型做推理：在处理简单指令路由时，我会先用一个 7B 甚至 3B 的小模型判断用户意图，只有当意图复杂（如需要多步推理）时，才切换到 72B+ 的大模型。这样能把路由阶段的成本降低 80%。
异步并行执行：如果 Agent 的几个子任务是相互独立的（比如同时去查三个不同渠道的数据），务必使用异步并发调用，而不是串行等待。

结语：从 Demo 到生产环境

AI Agent 自动化工作流已经不再是概念验证阶段。但要把 Demo 变成生产级系统，你需要关注工具调用的稳定性、参数的强类型约束，以及不可逆操作的审批机制。记住，Agent 不是要取代规则，而是要接管那些规则无法覆盖的模糊地带。如果你想深入了解特定框架的实现细节，推荐深入阅读 LangGraph 的官方文档关于「Subgraph」的部分。

用 AI Agent 构建自动化工作流：从原型到落地