用 Laravel 处理状态、队列、鉴权等企业逻辑,Python 微服务专注 LLM 编排与向量操作的分层架构实践。
当大多数开发者想到构建 AI Agent 时,他们会立刻跳入纯 Python 技术栈——FastAPI、LangChain 或 Autogen,跑在单一引擎上。
Python 在模型推理、embedding 生成和 LLM 编排方面无疑是王者。但当你需要将 AI Agent 打造成企业级产品——处理多租户认证、webhook 订阅、任务队列、计费和事务性数据库状态——用 Python 从头实现这些系统就是在浪费工程时间。
过去几年在构建复杂 ERP 和自主自动化工具的过程中(包括我在 SOFTDEFT 的 AI Bro 套件),我最终采用了一种混合架构,兼顾两者之长:
Laravel 负责状态管理、队列分发、认证、限速和面向客户端的 API。
Python 作为独立的高性能微服务,专门负责 LLM 编排、工具执行和向量操作。
以下是 Laravel 与 Python 协作构建弹性、生产级 AI Agent 的架构分解。
应用不应当在等待 LLM 思考 10–30 秒并调用外部 API 时阻塞 HTTP 请求,架构依赖异步任务队列和 webhook:
[ Client / Webhook ]
│
▼
[ Laravel Application ] ──(Pushes Job)──► [ Redis Queue ]
│
▼
[ Python Engine (FastAPI) ] ◄──(Executes)── [ Queue Worker / HTTP ]
│
├──► [ OpenAI / Claude / Local LLM ]
├──► [ Vector Database ]
└──► [ External Tools / APIs ]
│
[ Laravel Webhook Handler ] ◄──(Payload)──────┘
│
▼
[ Database & Client WebSockets ]
首先,我们需要跟踪 Agent 运行情况,但不能因此堵塞数据库或依赖易失的 Python 内存。在 Laravel 中,我们为 agent_tasks 定义一个迁移:
Schema::create('agent_tasks', function (Blueprint $table) {
$table->uuid('id')->primary();
$table->foreignId('user_id')->constrained()->cascadeOnDelete();
$table->string('agent_type'); // e.g., 'ecom_support', 'lead_qualifier'
$table->string('status')->default('pending'); // pending, processing, completed, failed
$table->json('input_payload');
$table->json('execution_log')->nullable();
$table->json('result')->nullable();
$table->timestamps();
});
当用户或外部 webhook 触发一个 Agent 操作时,Laravel 创建任务记录并分发一个异步任务:
namespace App\Jobs;
use App\Models\AgentTask;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Http;
class DispatchAgentTask implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public $tries = 3;
public $timeout = 120;
public function __construct(public AgentTask $task) {}
public function handle(): void
{
$this->task->update(['status' => 'processing']);
$response = Http::withHeaders([
'X-Internal-Secret' => config('services.agent_engine.secret'),
])->timeout(90)->post(config('services.agent_engine.url') . '/run-agent', [
'task_id' => $this->task->id,
'agent_type' => $this->task->agent_type,
'payload' => $this->task->input_payload,
'callback_url' => route('api.webhooks.agent-callback'),
]);
if ($response->failed()) {
$this->task->update(['status' => 'failed']);
$this->fail(new \Exception('Agent Engine Failed: ' . $response->body()));
}
}
}
在 Python 端,我们保持服务轻量且专注。使用 FastAPI 暴露专用端点,用标准 Python LLM 库处理决策逻辑。
以下是一个使用原生 function calling 的简化 Python runner:
import os
from fastapi import FastAPI, HTTPException, Header, BackgroundTasks
from pydantic import BaseModel
import httpx
from openai import OpenAI
app = FastAPI()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class AgentRequest(BaseModel):
task_id: str
agent_type: str
payload: dict
callback_url: str
def execute_agent_workflow(request: AgentRequest):
# Step 1: System Prompt Construction
system_prompt = (
"You are an autonomous business assistant. "
"Analyze the user request, call necessary tools, and arrive at a final resolution."
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": str(request.payload)}
]
# Step 2: Tool Definition
tools = [
{
"type": "function",
"function": {
"name": "check_inventory",
"description": "Check item stock level in the database",
"parameters": {
"type": "object",
"properties": {
"sku": {"type": "string"}
},
"required": ["sku"]
}
}
}
]
try:
# Step 3: LLM Inference & Loop
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
# Process tool calls or final output...
final_text = response.choices[0].message.content or "Task completed."
# Step 4: Callback to Laravel
httpx.post(request.callback_url, json={
"task_id": request.task_id,
"status": "completed",
"result": {"output": final_text}
}, timeout=10.0)
except Exception as e:
httpx.post(request.callback_url, json={
"task_id": request.task_id,
"status": "failed",
"error": str(e)
}, timeout=10.0)
@app.post("/run-agent")
async def run_agent(data: AgentRequest, background_tasks: BackgroundTasks, x_internal_secret: str = Header(None)):
if x_internal_secret != os.getenv("INTERNAL_ENGINE_SECRET"):
raise HTTPException(status_code=403, detail="Unauthorized")
# Run the heavy agent execution asynchronously in background
background_tasks.add_task(execute_agent_workflow, data)
return {"status": "accepted", "message": "Agent execution started."}
如果你计划在生产环境运行混合 Agent,请牢记以下三个边界情况:
严格的上下文边界:不要将原始数据库 dump 直接喂给 LLM prompt。先在 Laravel 端做过滤/摘要处理,以保持 token 成本和延迟可控。
幂等回调:网络超时难免发生。确保 Laravel 回调端点使用数据库事务,避免将 Agent 的操作应用两次。
优雅超时:始终将 Python 执行运行在 FastAPI 后台任务或专用 Celery 队列中,这样 Laravel 与 Python 之间的 HTTP 连接断开不会中断正在进行的模型调用。
将 Laravel 的后端稳定性与 Python 的 AI 能力结合,是一种简洁、可扩展的架构,适用于构建企业级 AI Agent。你在 PHP 中保持领域逻辑清晰,同时让 Python 做它最擅长的事。
Written by Faysal Ahmmed (Founder at SOFTDEFT). I specialize in architecting custom Laravel ERPs, enterprise web applications, and autonomous AI systems. Connect with me on faysaltanim.com