AI Agent 架构设计 · 运维手册
适用场景：设计生产级 AI Agent 系统，支撑自动化运维、智能问答、工单处理等场景。
1. Agent 架构模式
1.1 ReAct 模式（推理-行动）
适用场景：单步决策、信息查询、简单诊断
# ReAct Agent
from langchain.agents import create_react_agent
agent = create_react_agent(
llm=llm,
tools=tools,
prompt=react_prompt
)
# 执行流程
result = agent.invoke({
"input": "Nginx 502 错误，请帮我诊断"
})
1.2 Plan-and-Execute 模式
适用场景：多步骤运维任务（部署、升级、故障恢复）
from langchain.agents import create_plan_executor_agent
agent = create_plan_executor_agent(
llm=llm,
tools=tools,
planner=SequentialPlanner(), # 顺序规划器
executor=ConstrainedExecutor() # 约束执行器
)
1.3 Multi-Agent 模式
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Orchestrator│ │ Diagnostic │ │ Remediation │
│ (协调者) │ ──▶ │ (诊断者) │ ──▶ │ (修复者) │
└─────────────┘ └─────────────┘ └─────────────┘
▲ │ │
│ ▼ ▼
│ ┌─────────────┐ ┌─────────────┐
└──────────── │ Verifier │ ◀── │ Monitor │
│ (验证者) │ │ (监控者) │
└─────────────┘ └─────────────┘
适用场景：复杂运维流程（告警处理、变更管理）
2. Agent 核心组件
2.1 工具封装
from langchain.tools import Tool
import subprocess
def check_service(service: str) -> str:
"""检查服务状态"""
result = subprocess.run(
["systemctl", "is-active", service],
capture_output=True, text=True
)
return result.stdout.strip()
def query_metrics(metric: str) -> str:
"""查询指标"""
import requests
resp = requests.get(
f"http://prometheus:9090/api/v1/query?query={metric}"
)
return resp.text
tools = [
Tool(
name="check_service",
func=check_service,
description="检查 Linux 服务状态"
),
Tool(
name="query_metrics",
func=query_metrics,
description="查询 Prometheus 指标"
)
]
2.2 记忆系统
from langchain.memory import ChatMessageHistory
from langchain.vectorstores import Chroma
class AgentMemory:
def __init__(self):
self.short_term = {} # 会话记忆
self.long_term = Chroma(...) # 持久化记忆
def get_context(self, session_id: str, query: str) -> str:
"""获取上下文"""
# 短期记忆
history = self.short_term.get(session_id, [])
# 长期记忆（向量检索）
similar = self.long_term.similarity_search(query, k=3)
return f"历史对话: {history}\n相关知识: {similar}"
2.3 错误处理
from tenacity import retry, wait_exponential, stop_after_attempt
@retry(
wait=wait_exponential(multiplier=1, min=2, max=10),
stop=stop_after_attempt(3),
retry_if_exception_type(Exception)
)
def execute_with_retry(agent, input_text):
"""带重试的 Agent 执行"""
return agent.invoke({"input": input_text})
3. 生产级 Agent 设计
3.1 告警处理 Agent
class AlertAgent:
def __init__(self):
self.llm = ChatOllama(model="qwen2.5:7b")
self.tools = [check_service, query_metrics, read_logs]
self.memory = AgentMemory()
def process(self, alert: dict) -> dict:
# 1. 检索历史相似告警
context = self.memory.get_context(alert['session_id'], alert['description'])
# 2. 根因分析
analysis = self.analyze_root_cause(alert, context)
# 3. 生成修复计划
plan = self.generate_remediation_plan(analysis)
# 4. 执行或等待确认
if plan['severity'] in ['P0', 'P1']:
return self.ask_human_confirmation(plan)
else:
return self.execute_plan(plan)
3.2 巡检 Agent
class InspectionAgent:
def __init__(self):
self.tools = [check_cpu, check_memory, check_disk, check_services]
def run_inspection(self) -> dict:
results = {}
for tool in self.tools:
try:
results[tool.name] = tool.execute()
except Exception as e:
results[tool.name] = {"error": str(e)}
# 生成报告
report = self.generate_report(results)
return report
3.3 代码审查 Agent
class CodeReviewAgent:
def __init__(self):
self.tools = [read_file, run_linter, check_secrets]
def review(self, pr_url: str) -> dict:
# 1. 获取 PR 信息
pr = self.fetch_pr(pr_url)
# 2. 静态分析
issues = []
for tool in self.tools:
issues.extend(tool.analyze(pr))
# 3. 生成审查报告
report = self.format_review(pr, issues)
return report
4. 人机协作设计
4.1 确认节点
HIGH_RISK_OPERATIONS = [
"systemctl restart",
"docker rm -f",
"kubectl delete",
"rm -rf",
"iptables -F"
]
def should_ask_confirmation(operation: str) -> bool:
"""判断是否需要人工确认"""
for risky in HIGH_RISK_OPERATIONS:
if risky in operation:
return True
return False
4.2 审批流程
class ApprovalWorkflow:
def __init__(self):
self.pending_actions = {}
def submit(self, action: dict, requester: str) -> str:
"""提交审批"""
action_id = str(uuid.uuid4())[:8]
self.pending_actions[action_id] = {
"action": action,
"requester": requester,
"status": "pending"
}
return action_id
def approve(self, action_id: str, approver: str) -> bool:
"""审批通过"""
if action_id in self.pending_actions:
self.pending_actions[action_id]["status"] = "approved"
self.pending_actions[action_id]["approver"] = approver
return True
return False
5. 性能优化
5.1 缓存策略
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_tool_call(tool_name: str, args: tuple) -> str:
"""缓存工具调用结果"""
return execute_tool(tool_name, args)
5.2 并行执行
import asyncio
async def parallel_execution(tasks: list) -> list:
"""并行执行多个工具调用"""
return await asyncio.gather(*[
asyncio.to_thread(execute_tool, task['name'], task['args'])
for task in tasks
])
6. 监控与评估
6.1 Agent 指标
| 指标 | 计算方法 | 健康阈值 |
|---|---|---|
| 成功率 | 成功次数/总次数 | > 90% |
| 平均迭代数 | 工具调用次数 | < 5 |
| 完成时间 | P99 延迟 | < 30s |
| 人工干预率 | 需确认次数/总数 | < 20% |
6.2 日志记录
import logging
logger = logging.getLogger("agent")
def log_agent_action(agent_name: str, action: dict, result: dict):
logger.info({
"event": "agent_action",
"agent": agent_name,
"action": action,
"result": result,
"timestamp": datetime.now().isoformat()
})
7. 常见问题排查
问题 1：Agent 循环执行
症状：工具调用次数超过限制
解决： - 设置 max_iterations=5 - 检测重复状态 - 添加终止条件
问题 2：幻觉导致错误操作
症状：Agent 调用不存在的工具
解决： - 增加工具描述 - 使用 function calling - 添加输出验证
问题 3：上下文过长
症状：Token 超限或响应变慢
解决： - 摘要历史对话 - 使用长上下文模型 - 分块处理