.png)
这是关于AI智能体系列文章的第三篇。在前一篇文章中，我展示了大模型（LLMs）如何使用工具与现实世界系统互动，无需具体指令就能解决问题。虽然这种灵活性很强大，但它可能以控制力和可靠性为代价。在这里，我将讨论如何通过LLM工作流程来平衡这两种特性。我会先总体介绍如何设计这类系统，最后给出一个实施人工虚拟助手的具体例子。
所谓工作流程，就是产生预期结果的一系列步骤。比如，我发送邮件的流程通常如下：
有人给我发邮件。
我阅读邮件。
如果重要就回复，不重要就删除。
工作流程是构建自动化的核心，让我们能把任务交给计算机处理，从而要么省去亲自动手的麻烦，要么以远超人类的速度完成。
过去要实现这一点，我们需要（非常细致地）定义流程中的每一步，并将这些步骤转化为计算机代码。然而如今，我们给计算机的指令已不必如此精确。
LLM工作流程
大模型（LLMs）提供了一种让计算机按照我们意图行事的新方法。我们不必编写代码，只需用自然语言向LLMs发出指令即可。
这就引出了所谓的自主决策工作流程，本质上就是包含LLM的工作流程。Ng大神创造了自主智能体（agentic）这一术语，用来描述具有一定自主能力（即执行自主行动的能力）的系统，通常由LLM驱动
设计模式
使用LLM构建工作流程时，会涌现出一些常见的设计模式。Anthropic发布的一篇博客对此作了很好的总结。以下是简要回顾。
链式模式 = 一系列基于LLM的步骤
路由模式 = 使用LLM将输入分配给不同的流程
并行处理：分片执行 = 同时执行多个步骤以降低延迟
并行模式：投票表决 = 并行执行相同步骤多次以提高性能
协调器-工作者模式 = 使用LLM分解任务并决定执行哪些步骤
评估-优化模式 = 评估LLM的输出并循环给予反馈
协调器-工作机模式
协调器-工作机模式与前面几种模式有一个关键区别：它的步骤不是按预定义流程执行，而是由LLM动态决定。
这一模式可通过两种方式实现：第一种给予开发者更多控制权，第二种则赋予LLM更大自主权。
任务交接= 协调器将信息传递给其他LLM，由它们完成任务
将LLM作为工具使用= 协调器以工具调用的方式使用LLM，并自行整合信息
注：这两种方法并非互斥
评估-优化模式
相对于封闭式工作流的更进一步，是在开放式循环中运行LLM，这个循环会持续执行直到LLM的输出满足某些标准。在这些情况下，定义这些标准（即评估）是关键考虑因素。
通常来说，有三种实现方式：
基于规则的评估= 通过代码验证LLM的输出
基于机器学习的评估= 使用机器学习模型预测任务是否充分完成
用LLM作为评判者= 使用LLM来评估输出质量
LLM并非万能解药
虽然上述设计模式主要围绕LLM展开，但在用它们构建解决方案时，了解其优势与局限至关重要。例如：LLM擅长处理不可预测的用户输入，但若用于从简历中提取关键词，则成本高昂且不可靠。
对此，我认为有个实用框架可将软件分为三种类型。开发者需要根据问题约束条件，判断何种类型最适合工作流中的特定组件。
示例：人工虚拟助手(AVA)
让我们通过一个具体例子来看看这些概念的实际应用。在这里，我将使用OpenAI的Agents SDK实现一个简单的LLM工作流，为我执行任意的行政任务。
最终系统将能够处理各种请求，例如检索联系人信息和直接在我的Email账户中起草邮件。
系统设计
该工作流包含2个关键组件：规划器代理和执行器代理。它们都是配备了工具的LLM。
规划器具有只读工具。它的职责是收集信息并为执行器编写详细指令。如果不需要任何操作，它会输出一个终止工作流的二进制标志。
执行器的职责是执行“写入”操作。它的工具允许更新本地目录中的文件并向我的Gmail账户起草邮件。
将系统拆分为两个专门的LLM组件的一个关键优势是：它能处理比单一组件更复杂的任务。此外，通过分离读取和写入功能，我们降低了系统执行不良操作（例如删除重要信息）的风险。
创建智能体
我使用OpenAI的Agents SDK来实现规划器和执行器。为了确保规划器能可靠地输出用户备注、执行器指令和一个二进制exec_required标志，我们可以通过Pydantic强制模型使用结构化输出。
from pydantic import BaseModel
# defining structured output for Planner
class PlannerOutput(BaseModel):
exec_required: bool
exec_inst: str
user_note: str
创建规划器智能体时我们可以使用这个类，以确保其输出符合此格式。
from agents import Agent, ModelSettings
from tools.functions import read_instructions # custom function
from tools.for_agents import * # custom agent tools
# import openai sk and other environment variables from .env file
from dotenv import load_dotenv
load_dotenv()
# reading instructions from seperate files
planner_instructions = read_instructions("ava.md")
+ read_instructions("planner.md")
# creating planner agent
planner_agent = Agent(
name="Planner Agent",
instructions=planner_instructions,
tools=[read_dir_struct, read_file_contents],
model_settings=ModelSettings(temperature=0.5),
output_type=PlannerOutput,
)
由于规划器的指令较长，这里不做展示，想要源码的朋友们可以私信。该规划器被赋予两个工具，分别用于读取文件夹目录结构和文件内容，这些工具在此文件中定义。
我们同样可以创建一个执行器代理。不过不需要为它定义特殊的结构化输出，并会为其提供额外的工具来写邮件和覆盖文件。
# read instructions
executor_instructions = read_instructions("ava.md")
+ read_instructions("executor.md")
# Executor Agent
executor_agent = Agent(
name="Executor Agent",
instructions=executor_instructions,
tools=[read_dir_struct, read_file_contents,
write_email_draft, overwrite_existing_file],
model_settings=ModelSettings(temperature=0),
)
OAuth 授权
为了能让系统与我的Gmail账户交互，我需要完成一些前期准备工作，具体步骤可以了解如何获取gmail授权。
获取授权后，我们就可以在主脚本中调用该函数。
from tools.gmail import get_gmail_service
# check if token.json exists and initialize Gmail service if needed
if not os.path.exists(os.getenv("GOOGLE_TOKEN_PATH")):
print("--- Token not found, starting authentication process ---")
try:
get_gmail_service() # This will automatically handle the auth flow
print("--- Authentication successful ---")
except Exception as e:
print(f"--- Authentication failed: {str(e)} ---")
return
运行工作流
要运行这个工作流，我们需要创建一个异步函数，这样代理(agents)就可以从OpenAI的API流式传输tokens（通过这种方式，我们能实时查看代理在做什么并展示给用户）。我把这个函数称为main()。
from agents import Runner
async def main():
# read request from request.txt
with open("request.txt", "r") as file:
request = file.read()
# run planner agent
result = Runner.run_streamed(
planner_agent,
request,
)
# handle stream events
print("--- Analyzing ---")
await handle_stream_events(result)
# print planner agent's output
print()
print("--- Planner Agent Output ---")
print(result.final_output.user_note)
print()
# check if execution is required
if result.final_output.exec_required:
result = Runner.run_streamed(
executor_agent,
result.final_output.exec_inst
)
print("--- Executing ---")
await handle_stream_events(result)
print()
print("--- Executor Agent Output ---")
print(result.final_output)
print()
handle_stream_events() 是一个自定义函数，它会打印诸如LLM token输出和工具调用之类的信息。这个函数并非我们讨论的核心内容，但感兴趣的读者可以私信获取全部代码
演示
现在我们可以运行这个工作流程了！系统可以访问email-guides文件夹和directory.csv文件，从而帮助撰写特定格式的邮件并了解收件人是谁。下面是我让系统写一封"电话跟进"邮件的示例。
在这里使用带有通用工具的LLMs的好处是系统可以执行各种各样的其他任务，例如：在directory.csv中添加/删除联系人、将新的邮件模板添加到email-guides文件夹、建议在新联系人和现有联系人之间进行三方介绍，以及无数其他功能。
接下来的内容
与传统软件系统相比，LLM工作流让我们能创建更灵活的自动化方案。本文我们回顾了常见设计模式，然后实现了一个由2个配备工具的专用LLM组成的简单智能工作流。
虽然闭合式工作流能在使用LLM构建时兼顾确定性和灵活性，但某些应用场景更适合完全开放式地调用LLM。这就引出了"循环中的LLM"概念，这将是我们系列下一篇文章的重点。