企业官网建设流程全解析

现在的AI应用, 已经不再是那种只是你发出一个请求, 它就给你一个响应的简单系统了。

如果你已经完成了演示的步骤, 那么接下来的事情就是, AI智能体开始启动了。

当那种情形到来的时候,所谓的同步执行流程就会陷入崩溃的状态。

在这篇文章里, 我们详细地解释一下, 究竟应该怎样去使用, 以及怎样在特定的环境里, 针对AI智能体去构建一个既能够扩展又能容纳错误的发生, 也就是具备容错性的这样一个执行层次的系统架构。

1、在人工智能智能体系统中, 有关执行的难题。

基本上说, 人工智能应用的具体样子, 往往就是呈现如下这种模样:

用户 → LLM → 响应

但实际上, 智能体体系迅速地变化成了更加错综复杂的事物:

用户

↓

API

↓

任务队列

↓

AI智能体

├─ LLM推理

├─ 工具执行

├─ 外部API

├─ 重试和回退

└─ 结果合成

↓

结果存储

这引入了一些问题, 而且这些问题是不少的。

为了保证处理工作的可靠性, 执行部分必须被单独剥离出来, 从而实现与请求处理环节的相互脱节。

2、为什么要使用AI智能体呢?

它是一个分布式的任务队列系统, 专门针对这类问题进行了设计。

针对人工智能这一类体系结构, 它能够为使用者所提供的是:

不要把AI智能体简单地看成是用来调用的一些函数, 反而应该把它们看作是作业, 这种看法在规模上算是比较好的抽象。

在常见的生产配置过程中, 通常所使用的方法包括:

（代理）

Redis（结果后端）

这种分离的形态, 使得每一个独立的组件, 都能够去实现各自独立进行的扩展操作。

3、在后台, 人工智能智能体执行了第三点一的这一个场景。

当用户使用触发条件来启动这个AI智能体的时候, 该智能体就会执行相应的操作。

在这个过程被执行的时候, 不应该将API请求的流程堵住, 从而保证能够正常响应。

相反的情况是, 应用程序编程接口会进行作业的入队操作。

3.关于该项目的基本组织架构。

ai_agent_system/

├── celery_app.py

├── tasks.py

├── agent.py

├── app.py

└── requirements.txt

3.3 配置

.py

from celery import Celery

celery = Celery(

"ai_agent_tasks",

broker="amqp://guest:guest@localhost:5672//",

backend="redis://localhost:6379/0"

)

celery.conf.update(

task_serializer="json",

accept_content=["json"],

result_serializer="json",

timezone="UTC",

enable_utc=True,

)

这个设置比较简单, 也比较清楚, 并且很适合在正式生产的环节里面使用。

3.4 AI智能体逻辑

agent.py

import time

def run_ai_agent(prompt: str) -> dict:

# 第1步：推理

time.sleep(2)

analysis = f"分析输入：{prompt}"

# 第2步：工具执行

time.sleep(2)

tool_output = "外部数据已获取"

# 第3步：最终合成

time.sleep(2)

final_response = f"{prompt}的最终输出"

return {

"analysis": analysis,

"tool_output": tool_output,

"final_response": final_response

}

在我们现在所使用的实际系统里头, 这个函数有可能出现的情况是:

3.关于任务定义的这一章节, 具体内容为编号五。

tasks.py

from celery_app import celery

from agent import run_ai_agent

@celery.task(bind=True, max_retries=3)

def execute_ai_agent(self, prompt: str):

try:

return run_ai_agent(prompt)

except Exception as exc:

raise self.retry(exc=exc, countdown=5)

在AI系统里边, 重试这个动作是非常重要的一个环节。因为失败这一情况是属于预期之中会发生的常态, 并不是什么特殊的例外情况。

3.在第六个步骤当中, 通过API进行触发执行。

app.py

from tasks import execute_ai_agent

def handle_user_request(prompt: str):

task = execute_ai_agent.delay(prompt)

return {

"task_id": task.id,

"status": "智能体执行已启动"

}

这种模式可以允许API保持一种相应的状态, 也就是说, 不管智能体运行需要花费多长时间, 这种响应的状态都不会消失。

3.当需要进行任务调度的时候, 就会运行工作线程。

celery -A celery_app.celery worker --loglevel=info

因为智能体的负载增加了, 所以工作线程可以独立进行扩展。

3.第8部分, 获取任务的最终执行结果。

from celery.result import AsyncResult

from celery_app import celery

result = AsyncResult(task_id, app=celery)

if result.ready():

print(result.result)

这种情况所对应的, 通常是存在如下这些普遍性的规律模式。

4、生产考虑

在处理人工智能智能体的时候。

在AI系统里头, 绝大多数出现问题的地方是在那个编排的过程, 而并不在那个模型进行推理计算的地方。

5、结束语

AI智能体, 这个事物, 它在本质上面而言, 属于异步的类型。

它们会进行推理操作, 会耐心地等待时机, 会尝试重新执行任务, 并且还会和那些不太稳定的系统进行交互。如果把它们的运作方式简单地当成是普通的函数调用一样来处理的话, 那么最终的架构设计是非常容易出现问题的。

、和Redis提供了稳定的执行层, 从而允许AI智能体在真实世界条件下可以可靠地运行。

随着人工智能系统的自主化程度逐渐提高, 并且它能够进行长时间不间断的运行, 这种基础设施变得具有基础性重要意义, 它已经不再是可有可无的东西了。