Synchronous vs. Asynchronous Agent Execution: Architecture Patterns for Production

同步与异步代理执行：面向生产的架构模式 - MachineLearningMastery.com

同步与异步代理执行：面向生产的架构模式

By

Iván Palomares Carrascosa

on

2026年10月6日

in

人工智能

0

分享

文章

在本文中，你将学习同步和异步执行模式在架构上的差异，以及在将基于大语言模型（LLM）的代理部署到生产环境时如何在两者之间进行选择。

我们将涵盖的主题包括：

• 同步“等待观察”执行模式的工作原理，以及其适用的生产场景。
• 异步、事件驱动的“发送即忘记”模式如何将任务提交与任务完成解耦，以处理长期运行的代理工作流。
• 基于任务复杂度、延迟容忍度和基础设施需求，选择两种模式的实用指南。

引言

如今，借助层出不穷的高级库和支撑框架，构建本地Python脚本实现基于LLM的代理循环调用多个工具变得越来越容易。然而，代理AI领域存在严重的“部署差距”，需要进一步关注。

从生产角度来看，现实中的代理工作流包含复杂的依赖关系、多步骤推理循环和API延迟。未能设计出符合这些现实的架构会导致系统超时、用户请求丢失和内存泄漏。两种核心架构模式——同步和异步执行——能够弥合这些差距，并向标准分布式系统范式靠拢。本文从代理执行的角度探讨这两种模式，通过两个易于运行的轻量级笔记本说明其原理，这些笔记本的底层逻辑可直接转化为生产就绪的基于代理的应用程序。

同步代理执行：“等待观察”

同步代理执行模式与经典的HTTP请求-响应方法非常相似。它最适合需要即时反馈、任务间顺序依赖或高效数据检索的场景，例如标准的检索增强生成（RAG）管道。

其工作原理如下：外部调用者提交提示后，执行线程会被阻塞，直到代理完成整个思考链过程和工具使用。

虽然这种模式简单有效，但与高级代理结合时存在明显局限且非常脆弱。AWS等云基础设施提供商的API网关默认有29秒的超时限制。对于需要45秒来规划、搜索网络并生成响应的代理，标准同步管道会导致连接中断，造成令牌浪费和进度丢失。

以下是翻译后的 Markdown 内容：

以下代码展示了在 Google Colab 中可运行的示例，用于模拟同步代理执行模式。它首先定义了一个函数 mock_llm_call(prompt)，该函数模拟代理调用 LLM 的过程，但不实际使用 LLM。它接受一个提示作为输入，并模拟少量网络延迟，类似于调用真实 LLM 的效果。随后，简单的条件逻辑检查提示中是否包含单词 "search"。如果包含，函数会模拟执行网络搜索的工具调用。否则，它会模拟 LLM 生成最终答案，表示模型已处理请求并生成直接响应。在这两种情况下，函数都会返回一个描述操作及其相关元数据的字典。

import time

def mock_llm_call(prompt):
    """一个模拟的免费 LLM 调用（此处未实际使用 LLM），用于演示无需 API 密钥的延迟效果。"""
    time.sleep(1)  # 模拟网络延迟
    # 检查是否是需要搜索的初始请求
    if "research" in prompt.lower():
        return {"action": "tool_call", "tool": "web_search", "query": "latest agent news"}
    return {"action": "final_answer", "text": "Found the data! Agents are scaling."}

1

2

3

4

5

6

7

8

9

10

import

time

def

mock_llm_call

(

prompt

)

:

""

"A simulated, free LLM call (with no actual LLM used here) to demonstrate latency without API keys."

.

sleep

Simulate network latency

Checking if this is the initial request requiring a search

if

"research"

lower

return

{

"action"

"tool_call"

,

"tool"

"web_search"

"query"

"latest agent news"

}

"final_answer"

"text"

"Found the data! Agents are scaling."

同时，还需要一个调用前一个函数的总体代理函数。我们将这个函数称为 synchronous_agent(query)。它模拟了一个代理在多个步骤中处理查询的行为。在每一步中，代理都会调用 mock_llm_call()。如果模拟的 LLM 发出工具调用信号，代理会模拟工具执行并调整下一步的查询。如果 LLM 返回最终响应，代理将结束执行并返回该响应作为最终结果。

def synchronous_agent(query):
    print(f"[Sync API] Blocking thread to process: '{query}'")
    max_steps = 3
    for step in range(max_steps):
        print(f" -> Agent Step {step+1}: Thinking...")
        response = mock_llm_call(f"{query} (step {step})")
        if response["action"] == "final_answer":
            print(f"[Sync API] Finished! Returning payload to client.")
            return response['text']
        else:
            print(f" -> Executing Tool: {response['tool']}")
            # 更新查询以模拟注入工具的结果，避免触发关键词
            query = "Tool output: 'Agents require async architecture for scale.' Summarize this."
    return "Agent failed to complete in time."

11

12

13

14

15

16

17

18

synchronous_agent

query

print

f

"[Sync API] Blocking thread to process: '{query}'"

max_steps

=

for

step

range

" -> Agent Step {step+1}: Thinking..."

response

"{query} (step {step})"

[

]

==

"[Sync API] Finished! Returning payload to client."

'text'

else

" -> Executing Tool: {response['tool']}"

Updating query to simulate injecting the tool's result, avoiding trigger words

"Tool output: 'Agents require async architecture for scale.' Summarize this."

"Agent failed to complete in time."

让我们来试一下：

# 如何在 Colab 笔记本中运行代理：
result = synchronous_agent("Research AI agent patterns")
print(f"Result: {result}")

How to run the agent in a Colab notebook:

/think

"研究 AI 代理模式"

"结果：{result}"

结果：

[同步 API] 阻塞线程处理：'研究 AI 代理模式' -> 代理步骤 1：思考... -> 执行工具：web_search -> 代理步骤 2：思考... [同步 API] 完成！将负载返回给客户端。结果：找到数据了！代理正在扩展。

同步

API

阻塞

线程

处理

'Research AI agent patterns'

->

代理

思考

执行

工具

web_search

完成

！

返回

负载

客户端

找到

数据

代理

正在

扩展

我们刚刚看到了同步循环在实际中的工作方式。上面的代码模拟了一个代理执行推理步骤、调用工具，并在后续步骤中返回最终答案的过程。

异步、事件驱动执行：「即发即忘」

在处理复杂任务（如重构代码库、多代理辩论或需要频繁 HITL（人机协作）审批的工作流）时，需要转向异步执行模式。

在此模式下，任务提交和任务完成完全解耦。一旦客户端触发代理执行任务，系统会创建并返回一个 job_id，将任务移入队列。后台工作者随后从队列中获取任务并独立处理。这样，代理可以运行数小时而不会阻塞用户界面。任务状态会频繁检查点，以检测和恢复节点崩溃等问题，使代理能够从中断处继续执行。

此模式的缺点是需要更强大的基础设施，包括消息代理（如 RabbitMQ 和 Redis），以及适合建模工作者节点状态的数据库（如 PostgreSQL、MongoDB）。

让我们通过代码示例来理解异步模式。此处的关键构建模块是 Python 的 asyncio 库，其中“客户端”发送任务、接收确认，任务在后台处理：非常适合那些否则会阻塞部署应用程序用户界面的耗时任务。

agent_worker() 函数充当后台工作者节点，异步处理队列中的任务，模拟长时间运行的工作步骤并在数据库中更新状态。

import asyncio import uuid # 内存队列模拟 Redis/Celery（免费版） task_queue = asyncio.Queue() # 内存数据库模拟持久化状态检查点存储 database = {} async def agent_worker(): """独立处理长时间运行代理任务的后台工作者。""" while True: task = await task_queue.get() task_id = task['id'] print(f"\n[Worker] 已获取任务 {task_id}") # 模拟长时间运行的多步骤代理思考过程 database[task_id] = "运行步骤 1（规划）..." await asyncio.sleep(1.5) database[task_id] = "运行步骤 2（执行工具）..." await asyncio.sleep(1.5) # 保存最终状态（检查点） database[task_id] = "完成：生成了全面报告。" print(f"[Worker] 任务 {task_id} 完成。状态已保存至数据库。") task_queue.task_done()

19

20

21

22

23

24

25

26

asyncio

uuid

内存队列模拟 Redis/Celery（免费版）

task_queue

Queue

内存数据库模拟持久化状态检查点存储

database

async

agent_worker

"独立处理长时间运行代理任务的后台工作者。"

while

True

task

await

get

task_id

'id'

"\n[Worker] 已获取任务 {task_id}"

模拟长时间运行的多步骤代理思考过程

"正在执行步骤 1（规划）..."

1.5

"正在执行步骤 2（执行工具）..."

保存最终状态（检查点）

"完成：已生成综合报告。"

"[Worker] 任务 {task_id} 完成。状态已保存到数据库。"

task_done

接下来是 submit_job(prompt) 函数。该函数作为 API 接口接收任务，为其分配 ID 并加入队列，立即向客户端返回任务 ID 而不等待任务完成。

async def submit_job(prompt): """API 层：将任务提交到队列并立即返回。""" task_id = str(uuid.uuid4())[:8] await task_queue.put({"id": task_id, "prompt": prompt}) database[task_id] = "Pending" return task_id

submit_job

"API 层：将任务提交到队列并立即返回。"

str

uuid4

put

"id"

"prompt"

"Pending"

最后一个函数 main() 负责协调整个基于代理的系统：在后台启动工作节点，模拟客户端提交任务，并定期监控任务状态直至完成。

async def main(): # 1. 启动后台工作节点（作为我们的消费者舰队） worker = asyncio.create_task(agent_worker()) # 2. 客户端提交请求（API 不阻塞！） print("[API] 提交繁重任务...") job_id = await submit_job("撰写一份全面的多代理市场报告") print(f"[API] 成功！连接已关闭。返回任务 ID：{job_id}\n") # 3. 客户端定期检查状态（模拟前端轮询/网络钩子） for _ in range(4): print(f" [客户端轮询] 任务 {job_id} 状态：{database[job_id]}") await asyncio.sleep(1) # 为笔记本安全性清理无限工作节点 worker.cancel() # 如何在 Google Colab 笔记本中原生运行（已存在运行中的事件循环）： await main()

main

1. 启动后台工作节点（作为我们的消费者舰队）

worker

create_task

2. 客户端提交请求（API 不阻塞！）

"[API] 提交繁重任务..."

job_id

"撰写一份全面的多代理市场报告"

"[API] 成功！连接已关闭。返回任务 ID：{job_id}\n"

3. 客户端定期检查状态（模拟前端轮询/网络钩子）

_

" [客户端轮询] 任务 {job_id} 状态：{database[job_id]}"

为笔记本安全性清理无限工作节点

cancel

如何在 Google Colab 笔记本中原生运行（已存在运行中的事件循环）：

[API] 提交繁重任务... [API] 成功！连接已关闭。返回任务 ID：46f0c47a [客户端轮询] 任务 46f0c47a 状态：Pending [Worker] 接收到任务 46f0c47a [客户端轮询] 任务 46f0c47a 状态：执行步骤 1（规划）... [客户端轮询] 任务 46f0c47a 状态：执行步骤 2（执行工具）... [Worker] 任务 46f0c47a 完成。状态已保存到数据库。 [客户端轮询] 任务 46f0c47a 状态：完成：已生成综合报告。

提交

繁重

成功

连接

关闭

任务

ID

返回

46f0c47a

轮询

状态

待处理

接收到

执行

规划

工具

状态

保存

数据库

完成

综合

报告

生成

结论：何时使用哪种方式

一般来说，在处理问答等需要快速响应的任务时，应从轻量级的同步架构开始。随着代理应用功能的增强，通常需要转向异步执行，因为处理的任务耗时会更长。异步系统对基础设施（如数据库、队列、消息代理）提出了更高要求，但它们在应对脆弱的超时机制时具有更强的弹性，而这种弹性最终是将复杂代理工作流扩展到生产环境的关键。

更多相关内容

• 编码器-解码器 RNN 的实现模式…

• 将 AI 代理部署到生产环境：架构,…

• AI 代理的工具调用与代码执行对比：…

• Transformer 模型中的专家混合架构

• 用于罕见事件时间序列的 LSTM 模型架构…

• Python 中并发运行代理的 7 种异步模式

/.entry /think