Agent与大模型协同架构：从原理到工程实践

1次阅读

共计 2285 个字符，预计需要花费 6 分钟才能阅读完成。

在传统的 API 调用模式下，系统需要预先定义好所有的接口和数据格式，这在处理复杂工作流时存在明显局限性。比如订单处理系统，当涉及库存检查、支付验证、物流调度等多个步骤时，传统方式往往需要编写大量胶水代码来串联这些服务，系统会变得僵化且难以维护。

大模型的出现为解决这类问题提供了新思路。它们具备强大的自然语言理解和动态决策能力，可以根据上下文灵活调整处理流程。但这种能力也伴随着明显的缺陷：

API 调用延迟高（通常 500ms-5s）
输出结果不稳定
缺乏长期记忆能力
成本随调用次数线性增长

架构类型	QPS 上限	单次调用成本	可维护性	适用场景
纯大模型调用	10-50	高	低	简单问答、内容生成
纯 Agent 系统	1000+	低	中	固定流程任务
混合架构	100-300	中	高	复杂决策、动态工作流

@startuml
agent "任务 Agent" as agent
database "记忆存储" as memory
cloud "大模型 API" as llm

agent -> llm : 提交决策请求
llm --> agent : 返回 JSON 指令
agent -> memory : 保存上下文
agent --> memory : 读取历史
@enduml

from typing import Optional, Dict, Any
import asyncio
from pydantic import BaseModel

class AgentMemory(BaseModel):
    current_state: str
    history: list[dict]

class BaseAgent:
    def __init__(self, llm_client):
        self.memory = AgentMemory(
            current_state="init",
            history=[])
        self.task_queue = asyncio.Queue()
        self.llm = llm_client

    async def process_task(self, input_data: dict) -> dict:
        """异步处理任务的主循环"""
        try:
            # 步骤 1：调用大模型获取决策
            llm_response = await self._call_llm(input_data)

            # 步骤 2：更新记忆状态
            self._update_memory(llm_response)

            # 步骤 3：执行具体动作
            return await self._execute_action(llm_response)
        except Exception as e:
            self._handle_error(e)
            raise

    async def _call_llm(self, prompt: dict) -> dict:
        """调用大模型 API（带重试机制）"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                return await self.llm.generate(
                    prompt=prompt,
                    temperature=0.7
                )
            except Exception:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(1)

异步任务队列 ：使用 asyncio.Queue 实现生产者 - 消费者模式
记忆模块 ：采用 Pydantic 模型确保类型安全
错误处理 ：三级重试机制（网络错误、API 限流、超时）
类型标注 ：所有方法都有完整的类型提示

令牌桶算法实现（每秒钟 5 个请求）
动态退避策略（指数级增长等待时间）
优先队列（VIP 任务插队）

# 使用 Redis 保存状态示例
import redis
import json

r = redis.Redis()

def save_agent_state(agent_id: str, state: dict):
    r.set(f"agent:{agent_id}",
        json.dumps(state),
        ex=86400  # 24 小时过期
    )

输入输出过滤（正则表达式匹配敏感词）
内存隔离（每个 Agent 独立进程）
网络访问白名单

from locust import HttpUser, task

class AgentUser(HttpUser):
    @task
    def test_flow(self):
        self.client.post("/agent", json={
            "task": "订单处理",
            "params": {...}
        })