有粉丝关心本公众号近期发生迁移，在此特别说明：迁移仅为运营主体更名，小编们将一如既往的努力奉献优质内容。因迁移致公众号降权，若您喜爱我们的文章，点赞转发便是对我们莫大的支持，感谢一路相伴！祝大家双旦快乐！

Anthropic公司最近的一篇《Build effective agents》引起了广泛关注。不过简单的翻译从来不是我们的风格，经过几天的酝酿，今天来为大家解析这篇文章，并尝试用PydanticAI来完美实现文章中的设计模式，完成后你将对Agent工作流与来自大名鼎鼎Pydantic团队的新工具PydanticAI都会有全新的认识。

本篇目录：

1. 区分Agent系统、Workflows、Agents
   

2. Workflows的5种基础模式

3. 实现Workflows模式之链接模式
   

通常，在提到的Agent（智能体）的时候你会想到：能够感知环境，自主规划，并使用工具来完成多步骤任务的AI应用。在这里，自主的多步任务规划是智能体的标志性特征，也是Agent区分类似RAG应用的核心能力。

不幸的是，即使当前LLM能力已经非常惊艳，这种自主的多步任务准确率在很多测试中仅能达到接近人类的50%。这种“黑盒”Agent的不确定性，在很多复杂场景特别是企业应用中是致命的。也因此出现了另一种形式的应用：遵循预定义工作流程但仍然以LLM调用为核心来完成多步骤任务的系统。一般我们也称其为Agent。

回忆下LangChain为什么会推出LangGraph，LlamaIndex会推出Workflows？

Anthropic对这两种类型的Agent进行了清晰的划分与定义：

• Agent Systems（Agent系统）：不管是完全自主，还是预定义工作流的Agent都归为Agent系统。

  
  

• Workflows（工作流）：使用预定义流程编排LLM与工具来完成任务。

  
  

• Agents（智能体）：由LLM自主规划其任务步骤并完成任务的系统。核心特点是能够观察上个步骤的执行结果，来推理后续步骤，直到任务完成。因此，所有的Agents具有统一的工作模式：

总结：Workflows与Agents的核心区别在于其任务步骤是否由LLM自主规划。

理解了这一点，也就能得出什么时候用workflows，什么时候用Agents：

• 足够简单的任务既不用Workflows，也不用Agents。

  比如：让LLM从网络搜索最新天气

• 需要极高的可预测性与可靠性的复杂流程任务使用Workflows。

  比如：一个带有自我反省机制的并行编程Agent应用

  
  

• 需要一定灵活性，或无法预测任务步骤时，可考虑使用Agents。

  比如：一个动态规划访问路径的WebUI Agent。

Anthropic文章的另一个关键部分是对构建工作流的模式进行了清晰的总结，详细描述了工作流的5种基础模式。在这5种基础模式上，结合必要的控制流程，可以组装更复杂的工作流。

需要注意的是，这5种基础模式的构建单元是增强型LLM（调用）：

总结：增强型LLM就是带有工具调用、记忆与检索能力的LLM。你甚至可以理解成缺乏规划能力但会使用工具的“还不够智能”的Agent。

下面介绍五种基础工作流模式：

1. 链接模式

也可以理解为顺序模式：多个LLM调用按输入输出串接起来执行。

【用例】搜索一篇文章，然后翻译成某种语言，最后发送邮件。

2. 路由模式

一个LLM调用作为路由器，将任务转发到后续不同的LLM调用。

【用例】将客户服务的请求按类型转给不同的后续LLM处理。

3. 并行模式

多个LLM调用并行处理，最后对处理结果进行汇聚后输出。

【用例】对一段代码从语法、逻辑、规范不同的角度做审查。

4. 编排者-工作者模式

类似于多智能体的Supervisor模式：一个管理LLM负责编排与分解出子任务，多个工作LLM负责子任务的完成，直到任务完成。

【用例】从不确定的多个数据源搜集信息，并形成综合报告。

5. 评估器-优化器模式

一个LLM负责处理任务，另一个LLM则负责评估并反馈。经过多次迭代后，直到任务完成。

【用例】翻译任务中一个LLM负责输出，一个LLM负责提出改进意见。

在实现Agent系统的方法上，Anthropic建议：

尽可能的直接使用LLM API，而非引入框架，因为框架会带来复杂性，并使得调试困难。

不过在这点上我们还是认为大部分时候引入框架的收益要大于代价，原因是：

• 简化开发工作量。应用越复杂，收益越大

• 通过抽象层灵活适应变化，保留更好的扩展性

• 调试复杂性也可以借助成熟的工程化平台

• 主流框架都开源，有丰富的社区支持

  
  

所以为了更具有通用性和简洁性，我们这里用轻量级的框架PydanticAI来实现Workflows的5种基础模式。

首先来实现链接（顺序）模式，这里的模拟用例是：

从网络搜索要求的内容，翻译成英文后，发送电子邮件。

实现要点是：

• 使用PydanticAI的Agent来模拟“增强型LLM“调用

• 为了更好演示“增强型LLM”，使用了工具

• 每个步骤可以配置不同的提示、模型、工具、结构化输出类型

下面直接给出代码，参考注释即可看懂：

import asyncio,pprint
from typing import List

from pydantic import BaseModel
from pydantic_ai import Agent,Tool
from pydantic_ai.models.openai import OpenAIModel
from langchain_community.tools.tavily_search import TavilySearchResults

#tool:搜索网络信息
def tool_search(query: str) -> str:
    """用于搜索网络信息"""

    search = TavilySearchResults(max_results=1)
    results = search.invoke(query)
    return "\n".join([r["content"] for r in results])

#tool:发送邮件(模拟)
def tool_sendmail(subject: str, body: str,recipient: str = None) -> str:
    """用于发送邮件"""
    
    print(f"开始发送邮件...")
    return f"Sent email to {recipient} with subject '{subject}' and body '{body}'"

#邮件模型，演示PydanticAI的自动类型转换
class Mail(BaseModel):
    subject: str
    recipient: str
    body: str

#链式调用
async def chain(input: str, steps: List[dict]) -> str:
    """按顺序链接多个LLM调用,在步骤之间传递结果。"""

    result = input
    for i, step in enumerate(steps, 1):

        #读取步骤配置
        prompt = step["prompt"]
        tools = [Tool(tool) for tool in step["tools"]]
        model_config = step.get("model_config", {})
        model = OpenAIModel(**model_config)
        result_type = step.get("result_type", str)

        #创建LLM调用
        agent = Agent(model, system_prompt=prompt, tools=tools, result_type=result_type)

        #运行LLM调用
        response = await agent.run(f"输入:\n{result}")
        result = response.data
        print(f"步骤 {i}: {result}\n\n")
    return result

#链处理步骤（每个步骤可以配置提示、工具、模型、返回类型）
data_processing_steps = [
        {
            "prompt": "从网络搜索与输入问题相关的信息",
            "tools": [tool_search],
            "model_config": {"model_name": "gpt-4o-mini"},
            "result_type": str
        },
        {
            "prompt": "将上述中文内容翻译成英文",
            "tools": [],
            "model_config": {"model_name": "gpt-4o-mini"},
            "result_type": str
        },
        {
            "prompt": "将内容发送邮件到test@example.com",
            "tools": [tool_sendmail],
            "model_config": {"model_name": "gpt-4o-mini"},
            "result_type": Mail
        }
    ]

用如下代码来测试：

initial_input = "明天北京的天气"

async def main():
    result = await chain(initial_input, data_processing_steps)
    
    print('\n')
    pprint.pprint(result)

if __name__ == "__main__":
    asyncio.run(main())

获得输出如下：

可以看到，程序按照设定的工作流步骤顺利完成输入任务，其中每个步骤都是一个“增强型LLM”的调用。

后续我们将接着实现其他4种模式，敬请期待。

为了帮助LLM开发人员更系统性与更深入的学习RAG应用，特别是企业级的RAG应用场景下，当前主流的优化方法与技术实现，我们编写了《基于大模型的RAG应用开发与优化 — 构建企业级LLM应用》这本长达500页的开发指南，与大家一起来深入到LLM应用开发的全新世界。

更多细节，点击如下链接了解

现在购，享50%折扣