AI Agent工作流的5种基础模式及其通用实现(PydanticAI)【上篇】

360影视 国产动漫 2025-04-10 05:36 2

摘要:Anthropic公司的《Build effective agents》引起了广泛关注。不过简单的翻译从来不是我们的风格,经过几天的酝酿,今天来为大家解析这篇文章,并尝试用PydanticAI来完美实现文章中的设计模式,完成后你将对Agent工作流与来自大名鼎

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

本篇目录:

区分Agent系统、Workflows、AgentsWorkflows的5种基础模式实现Workflows模式之链接模式

01

区分Agent系统、Workflows、Agents

通常,在提到的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。

02

Workflows的5种基础模式

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负责提出改进意见。

03

实现Workflows模式之链接模式

在实现Agent系统的方法上,Anthropic建议:

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

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

简化开发工作量。应用越复杂,收益越大通过抽象层灵活适应变化,保留更好的扩展性调试复杂性也可以借助成熟的工程化平台主流框架都开源,有丰富的社区支持

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

首先来实现链接(顺序)模式,这里的模拟用例是:

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

实现要点是:

使用PydanticAI的Agent来模拟“增强型LLM“调用为了更好演示“增强型LLM”,使用了工具每个步骤可以配置不同的提示、模型、工具、结构化输出类型

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

import asyncio,pprintfrom typing import Listfrom pydantic import BaseModelfrom pydantic_ai import Agent,Toolfrom pydantic_ai.models.openai import OpenAIModelfrom 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: strrecipient: strbody: str#链式调用async def chain(input: str, steps: List[dict]) -> str:"""按顺序链接多个LLM调用,在步骤之间传递结果。"""result = inputfor 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.dataprint(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深度应用,关注我不迷路

所有平台同号

来源:蜂宇同程e

相关推荐