3、从“数据流”到“意图流”：传统pipeline与Agent工作流的本质区别
好，咱们今天聊点硬核的。
很多朋友问我：“老师，Agent 到底牛在哪？不就是把几个模型串起来吗？”
嗯，这个问题问到了根上。如果你只是把几个模型串起来，那叫 pipeline，不叫 Agent。
我个人习惯把传统 pipeline 叫做“数据流”，把 Agent 叫做“意图流”。
这两者，本质上是两种完全不同的世界观。
3.1 传统pipeline：被动的“数据搬运工”
先说说传统 pipeline。你想想看，它像什么？
像一条流水线。原料从这头进去，经过 A 工序、B 工序、C 工序，最后出来成品。
每个环节都是死的。A 做完，必须交给 B。B 做完，必须交给 C。
数据在中间流，但没有任何一个环节会问：“我为什么要做这个？”
传统pipeline的核心特征：
- 确定性路径：输入 -> 处理A -> 处理B -> 输出，路径写死在代码里
- 无状态感知：每个模块只知道自己当前的数据，不知道上下文
- 无自我修正：如果A步骤错了，B和C只能将错就错
- 线性依赖：一个环节卡住，整个流程就停了
我在项目中遇到过最典型的例子：一个 NLP 文本分类 pipeline。
先分词，再词性标注，再实体识别，最后分类。
看起来挺完整对吧？但问题来了——如果分词阶段把一个专有名词切错了，后面的词性标注、实体识别全都会跟着错。
而且，没有任何一个环节会回头说：“喂，分词那哥们，你刚才切错了，重来一遍。”
这就是数据流的悲哀。数据在流，但没有人思考。
3.2 Agent工作流：主动的“意图执行者”
Agent 工作流就不一样了。
它不再是“数据流”，而是“意图流”。
什么意思？
Agent 的核心，是它有一个“意图”。它知道自己要干什么，然后根据这个意图，动态地决定下一步该做什么。
Agent工作流的核心特征：
- 动态路径：根据当前状态和意图，实时决策下一步
- 上下文感知：每个决策都基于完整的对话历史和任务状态
- 自我修正：发现错误可以回溯、重试、甚至改变策略
- 并行与异步：可以同时执行多个子任务，等待结果再聚合
说白了，Agent 像是一个有脑子的项目经理。它不会傻傻地按流程走，它会看情况。
比如，同样是做文本分类，Agent 会怎么做？
它会先问自己：“这个文本是什么领域的？如果是医疗领域，我需要调用医疗词典。如果是法律领域，我需要调用法律知识库。”
然后，它根据这个判断，动态地选择不同的处理路径。
如果中间发现某个实体识别结果置信度太低，它会主动去查一下外部知识库，或者换个模型再试一次。
3.3 一张图看懂区别
为了让你更直观地理解，我画了一张对比图。
你看，左边是传统 pipeline，一条直线走到黑。右边是 Agent，有分支、有汇聚、有反馈循环。
这就是本质区别。
3.4 代码层面的差异
光说理论不够，咱们上代码看看。
传统pipeline的代码长这样：
def traditional_pipeline(text):
# 步骤1：分词
tokens = tokenize(text)
# 步骤2：词性标注
pos_tags = pos_tag(tokens)
# 步骤3：实体识别
entities = ner(pos_tags)
# 步骤4：分类
result = classify(entities)
return result
# 调用
output = traditional_pipeline("今天天气真好")
# 如果tokenize出错了，后面全错，没人管Agent工作流的代码长这样：
class Agent:
def __init__(self):
self.tools = {
"tokenizer": Tokenizer(),
"pos_tagger": POSTagger(),
"ner": NER(),
"classifier": Classifier(),
"knowledge_base": KnowledgeBase()
}
self.memory = []
def execute(self, intent, text):
# 1. 解析意图
plan = self.parse_intent(intent)
# 2. 动态执行
for step in plan:
result = self.tools[step["tool"]].run(text)
# 3. 验证结果
if result.confidence < 0.7:
# 置信度低，查知识库
correction = self.tools["knowledge_base"].query(text)
result = self.retry_with_correction(step, correction)
# 4. 记录到记忆
self.memory.append(result)
# 5. 聚合输出
return self.aggregate(self.memory)
# 调用
agent = Agent()
output = agent.execute("分类并提取实体", "今天天气真好")
# 如果某步出错，Agent会自己修正看到了吗？
传统 pipeline 是“我告诉你做什么，你就做什么”。
Agent 是“我告诉你我要什么，你自己想办法”。
3.5 避坑指南
我曾经犯过的错：
刚开始做 Agent 时，我犯了一个典型的错误——把 Agent 做成了“带 if-else 的 pipeline”。
什么意思？就是虽然用了 Agent 框架，但本质上还是写死了路径：
- 如果输入是A，走路径1
- 如果输入是B，走路径2
这根本不是 Agent。这只是披着 Agent 外衣的 pipeline。
真正的 Agent，它的决策应该是基于“意图”和“当前状态”动态生成的，而不是基于预设的规则。
我的建议：
判断你的系统是 pipeline 还是 Agent，有一个简单的方法：
问自己一个问题——如果中间某一步失败了，系统会怎么做？
- 如果答案是“报错，然后停止”，那它是 pipeline。
- 如果答案是“尝试其他方法，或者回溯重试”，那它才是 Agent。
3.6 什么时候用pipeline，什么时候用Agent？
这里我要说一句大实话：不是所有场景都需要 Agent。
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 数据清洗、ETL | Pipeline | 路径固定，不需要决策 |
| 批量文本分类 | Pipeline | 输入输出明确，不需要上下文 |
| 复杂对话系统 | Agent | 需要理解意图、动态决策 |
| 多步骤任务编排 | Agent | 需要自我修正和上下文感知 |
| 简单API调用 | Pipeline | 杀鸡不用牛刀 |
嗯，这里要注意：不要为了用 Agent 而用 Agent。
如果你的业务逻辑就是固定的，用 pipeline 反而更稳定、更高效。
Agent 的优势在于“不确定性”和“复杂性”。如果你的场景很简单，强行上 Agent 只会增加系统的复杂度和延迟。
3.7 总结一下
从“数据流”到“意图流”，本质上是思维方式的转变。
传统 pipeline 是“我告诉你怎么做”，Agent 是“我告诉你要什么”。
前者关注的是“流程”，后者关注的是“目标”。
我个人觉得，这个转变是 AI 工程师从“工具使用者”到“系统架构师”的关键一步。
你不再只是把模型串起来，而是开始思考：系统如何理解意图？如何动态决策？如何自我修正？
这才是 Agent 思维的核心。
公众号:蓝海资料掘金营，微信deep3321