AI Agent 生产部署与工程化

生产部署涉及可靠性、可扩展性、成本控制等多个方面。

工程化实践确保系统稳定运行并便于维护。

生产部署架构

生产环境对系统有更高的要求。

需要考虑可靠性、可扩展性、监控告警等多个方面。

部署模式

单机部署

适合开发和测试环境。

简单易部署，但无法应对生产级流量。

分布式部署

适合生产环境，需要考虑多个方面。

负载均衡：分发请求到多个实例。

服务发现：动态管理实例列表。

状态管理：处理分布式状态一致性问题。

容错处理：单点故障不影响整体服务。

容器化部署

Dockerfile 示例

# 基于 Python 3.11 镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建非 root 用户（安全考虑）
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "agent.py"]

Kubernetes 部署

Kubernetes Deployment 配置

span style="color: green;">
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-agent
labels:
app: ai-agent
spec:
# 副本数
replicas: 3
selector:
matchLabels:
app: ai-agent
template:
metadata:
labels:
app: ai-agent
spec:
containers:
- name: agent
image: your-registry/ai-agent:latest
ports:
- containerPort: 8000
# 资源限制
resources:
limits:
memory: "2Gi"
cpu: "1"
requests:
memory: "1Gi"
cpu: "0.5"
# 健康检查
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: ai-agent-service
spec:
selector:
app: ai-agent
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

成本控制

Agent 系统的成本主要来自 LLM API 调用。

优化成本是生产部署的重要课题。

Token 优化策略

提示词压缩：精简提示词，减少不必要的文字。

上下文截断：只保留关键的上下文内容。

缓存响应：对相同或相似查询缓存响应。

代码实现

成本优化 Agent 实现

import hashlib
import json
from functools import lru_cache

class CostOptimizedAgent:
"""
成本优化的 Agent
通过缓存和压缩减少 API 调用成本
"""

def __init__(self, base_agent, cache, embedder=None):
# 基础 Agent
self.base_agent = base_agent
# 缓存存储
self.cache = cache
# 嵌入器（用于相似度缓存匹配）
self.embedder = embedder

def process(self, request):
"""
处理请求，包含成本优化逻辑
"""
# 生成请求指纹
request_fingerprint = self.generate_fingerprint(request)

# 检查精确缓存
cached_result = self.cache.get(request_fingerprint)
if cached_result:
return {
**cached_result,
"from_cache": True
}

# 如果有嵌入器，检查语义相似缓存
if self.embedder:
similar = self.find_similar_cached(request)
if similar:
return {
**similar,
"from_cache": True,
"similar_to": similar.get("request_id")
}

# 执行请求
result = self.base_agent.process(request)

# 缓存结果
self.cache.set(
request_fingerprint,
result,
ttl=3600 # 1 小时过期
)

return {
**result,
"from_cache": False
}

def generate_fingerprint(self, request):
"""
生成请求指纹
用于精确缓存匹配
"""
content = json.dumps(request, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()

def find_similar_cached(self, request, threshold=0.95):
"""
查找语义相似的缓存结果
使用嵌入向量的余弦相似度
"""
if not self.embedder:
return None

# 将请求向量化
request_vector = self.embedder.embed([str(request)])[0]

# 在缓存中搜索相似请求
best_match = None
best_similarity = 0

for cached in self.cache.get_all():
cached_vector = cached["vector"]
similarity = cosine_similarity(request_vector, cached_vector)

if similarity > best_similarity and similarity >= threshold:
best_similarity = similarity
best_match = cached

return best_match

class ResponseCache:
"""
响应缓存
简单的内存缓存实现
生产环境可使用 Redis
"""

def __init__(self):
self.cache = {}
self.timestamps = {}

def get(self, key):
"""获取缓存"""
if key in self.cache:
# 检查是否过期
if self.is_expired(key):
self.delete(key)
return None
return self.cache[key]
return None

def set(self, key, value, ttl=3600):
"""设置缓存"""
self.cache[key] = value
self.timestamps[key] = time.time() + ttl

def delete(self, key):
"""删除缓存"""
if key in self.cache:
del self.cache[key]
if key in self.timestamps:
del self.timestamps[key]

def is_expired(self, key):
"""检查是否过期"""
if key not in self.timestamps:
return True
return time.time() > self.timestamps[key]

def get_all(self):
"""获取所有缓存项"""
return list(self.cache.values())

def cosine_similarity(vec1, vec2):
"""计算余弦相似度"""
dot_product = sum(a * b for a, b in zip(vec1, vec2))
norm1 = math.sqrt(sum(a * a for a in vec1))
norm2 = math.sqrt(sum(b * b for b in vec2))
return dot_product / (norm1 * norm2)

流式与异步

流式响应

流式响应减少等待时间，提升用户体验。

不用等待完整响应，可以逐步显示生成内容。

代码实现

流式响应 Agent

import asyncio

class StreamingAgent:
"""
支持流式响应的 Agent
边生成边返回，减少等待时间
"""

def __init__(self, llm):
self.llm = llm

async def stream_generate(self, prompt):
"""
流式生成响应
使用 async generator 模式
:yield: 生成的文本片段
"""
# 启动异步生成任务
async for chunk in self.llm.stream_generate(prompt):
yield chunk

async def process_stream(self, request):
"""
处理流式请求
返回异步生成器
"""
prompt = self.build_prompt(request)

# 返回生成器
async def generate():
async for chunk in self.stream_generate(prompt):
yield chunk

return generate()

# 使用示例
async def main():
agent = StreamingAgent(llm)

# 启动流式生成
async for token in agent.stream_generate("解释量子计算"):
# 边生成边打印
print(token, end="", flush=True)
print() # 换行

# 运行
asyncio.run(main())

异步任务队列

对于长时间运行的任务，使用异步队列处理。

用户提交任务后立即返回，后台异步执行。

异步任务队列实现

import asyncio
from queue import Queue
from threading import Thread
import uuid

class AsyncAgent:
"""
异步 Agent
使用任务队列处理长时间运行的任务
"""

def __init__(self, agent, task_queue):
# 基础 Agent
self.agent = agent
# 任务队列
self.task_queue = task_queue
# 结果存储
self.results = {}
# 启动后台工作线程
self.worker_thread = Thread(target=self.process_queue, daemon=True)
self.worker_thread.start()

async def submit(self, task):
"""
提交任务
立即返回任务 ID
"""
# 生成唯一任务 ID
task_id = str(uuid.uuid4())

# 添加到队列
await self.task_queue.enqueue({
"id": task_id,
"task": task,
"status": "pending",
"created_at": time.time()
})

return task_id

async def get_result(self, task_id):
"""
获取任务结果
非阻塞式
"""
if task_id in self.results:
return self.results[task_id]

# 检查任务状态
status = await self.task_queue.get_status(task_id)

if status is None:
return None # 任务不存在

if status == "pending" or status == "processing":
return {
"status": status,
"result": None
}

return None

def process_queue(self):
"""
后台工作线程
从队列取出任务并执行
"""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

while True:
task = loop.run_until_complete(self.task_queue.dequeue())

if task:
# 更新状态为处理中
loop.run_until_complete(
self.task_queue.update_status(task["id"], "processing")
)

try:
# 执行任务
result = self.agent.process(task["task"])

# 保存结果
self.results[task["id"]] = {
"status": "completed",
"result": result,
"completed_at": time.time()
}

loop.run_until_complete(
self.task_queue.update_status(task["id"], "completed")
)
except Exception as e:
self.results[task["id"]] = {
"status": "failed",
"error": str(e)
}

loop.run_until_complete(
self.task_queue.update_status(task["id"], "failed")
)

class TaskQueue:
"""简单的任务队列实现"""

def __init__(self):
self.queue = Queue()
self.statuses = {}

async def enqueue(self, task):
"""添加任务到队列"""
self.queue.put(task)
self.statuses[task["id"]] = "pending"

async def dequeue(self):
"""取出一个任务"""
if not self.queue.empty():
return self.queue.get()
return None

async def get_status(self, task_id):
"""获取任务状态"""
return self.statuses.get(task_id)

async def update_status(self, task_id, status):
"""更新任务状态"""
self.statuses[task_id] = status

微服务化设计

将 Agent 系统拆分为多个独立服务，提高可维护性和可扩展性。

服务划分

服务	职责
API 网关	统一入口，认证限流，路由转发
Agent 服务	核心逻辑，推理规划
工具服务	外部 API 集成，工具调用
存储服务	知识库，状态管理，缓存
监控服务	日志收集，指标监控，告警

服务间通信

同步通信：HTTP/gRPC，用于实时请求响应。

异步通信：消息队列，用于耗时操作和事件通知。

章节小结

本章节介绍了生产部署与工程化的关键实践。

生产部署架构 从单机到分布式，逐步扩展。

容器化部署 使用 Docker 和 Kubernetes。

成本控制 通过缓存和 Token 优化降低成本。

流式与异步 提升用户体验和系统吞吐量。

微服务化设计 提高系统的可维护性和可扩展性。

生产部署需要综合考虑可靠性、性能、成本等多个方面。

建议从小规模开始，逐步增加复杂度。