生产部署：可靠性与安全

生产环境的挑战清单

超时与重试策略

import asyncio
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log
)
import logging
from langchain_openai import ChatOpenAI
import openai

logger = logging.getLogger(__name__)

# ── 带指数退避的重试装饰器 ────────────────────────────────
@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.APIConnectionError,
    )),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
async def call_llm_with_retry(messages):
    """带重试的 LLM 调用。自动处理限速和超时。"""
    return await llm.ainvoke(messages)

# ── Agent 级别的超时控制 ──────────────────────────────────
async def run_agent_with_timeout(user_input: str, timeout: int = 120):
    """运行 Agent，超过 timeout 秒则强制终止。"""
    try:
        result = await asyncio.wait_for(
            graph.ainvoke({"messages": [HumanMessage(content=user_input)]}),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        logger.error(f"Agent 超时（>{timeout}s），强制终止。input={user_input[:100]}")
        return {"error": "处理超时，请简化您的请求。"}

# ── 最大迭代次数保护 ──────────────────────────────────────
class SafeAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration: Annotated[int, operator.add]

def safe_router(state: SafeAgentState) -> str:
    if state["iteration"] >= 15:
        logger.warning(f"Agent 超过最大迭代次数 (15)")
        return "force_end"
    last = state["messages"][-1]
    return "tools" if last.tool_calls else "end"

def agent_node_safe(state: SafeAgentState):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response], "iteration": 1}  # 每次 +1

成本控制策略

from langchain_core.callbacks import BaseCallbackHandler
from dataclasses import dataclass, field
from threading import Lock
import time

# ── Token 用量追踪器 ──────────────────────────────────────
@dataclass
class TokenBudget:
    max_tokens_per_run: int = 50_000
    max_cost_per_run: float = 0.5   # 美元
    used_tokens: int = 0
    _lock: Lock = field(default_factory=Lock)

    # GPT-4o-mini 价格（2025年）
    INPUT_PRICE_PER_1K = 0.00015   # $0.15/1M tokens
    OUTPUT_PRICE_PER_1K = 0.0006   # $0.60/1M tokens

    def add_usage(self, input_tokens: int, output_tokens: int):
        with self._lock:
            self.used_tokens += input_tokens + output_tokens
            cost = (input_tokens * self.INPUT_PRICE_PER_1K / 1000 +
                    output_tokens * self.OUTPUT_PRICE_PER_1K / 1000)
            if self.used_tokens > self.max_tokens_per_run:
                raise RuntimeError(
                    f"Token 预算超限：已用 {self.used_tokens}/{self.max_tokens_per_run}"
                )

class CostGuardCallback(BaseCallbackHandler):
    def __init__(self, budget: TokenBudget):
        self.budget = budget

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        self.budget.add_usage(
            usage.get("prompt_tokens", 0),
            usage.get("completion_tokens", 0)
        )

# ── 模型降级策略（成本分层）──────────────────────────────
def get_model_for_task(task_type: str) -> ChatOpenAI:
    """根据任务复杂度选择合适的模型，平衡质量和成本。"""
    model_map = {
        "routing":    "gpt-4o-mini",   # 简单分类：便宜模型
        "retrieval":  "gpt-4o-mini",   # 检索决策：便宜模型
        "reasoning":  "gpt-4o",        # 复杂推理：高质量模型
        "generation": "gpt-4o",        # 最终生成：高质量模型
        "evaluation": "gpt-4o",        # 质量评估：高质量模型
    }
    model = model_map.get(task_type, "gpt-4o-mini")
    return ChatOpenAI(model=model, temperature=0)

提示注入防护

提示注入（Prompt Injection）是 Agent 最危险的安全漏洞之一：攻击者通过构造恶意输入，试图覆盖 Agent 的指令：

import re
from typing import Optional

# ── 输入净化 ──────────────────────────────────────────────
INJECTION_PATTERNS = [
    r"\[SYSTEM\]",
    r"ignore (all |previous |above )?instructions",
    r"you are now",
    r"(jailbreak|DAN|developer mode)",
    r"disregard (your |all |previous )?instructions",
]

def sanitize_user_input(user_input: str) -> Optional[str]:
    """检测并拒绝可疑的提示注入尝试。"""
    lower = user_input.lower()
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, lower, re.IGNORECASE):
            logger.warning(f"检测到注入尝试：{user_input[:100]}")
            return None  # 拒绝输入

    # 长度限制
    if len(user_input) > 10000:
        return user_input[:10000]

    return user_input

# ── 输出安全过滤 ──────────────────────────────────────────
class OutputSafetyFilter(BaseModel):
    is_safe: bool
    risk_category: Optional[str] = None
    safe_response: str

safety_llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(OutputSafetyFilter)

def filter_agent_output(raw_output: str) -> str:
    """对 Agent 输出进行安全检查。"""
    result = safety_llm.invoke([
        SystemMessage(content="""检查以下AI回复是否包含：
        - 个人身份信息（姓名/手机/地址/证件号）
        - 有害内容（暴力/违法/歧视）
        - 系统内部信息（API密钥/内部路径）
        如不安全，提供替代的安全回复。"""),
        HumanMessage(content=raw_output)
    ])
    if not result.is_safe:
        logger.warning(f"输出安全过滤触发，类别：{result.risk_category}")
        return result.safe_response
    return raw_output

监控告警体系

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# ── Prometheus 指标定义 ───────────────────────────────────
agent_runs_total = Counter(
    "agent_runs_total",
    "Agent 运行总次数",
    ["status", "agent_type"]   # status: success/failed/timeout
)
agent_duration = Histogram(
    "agent_duration_seconds",
    "Agent 运行耗时分布",
    buckets=[1, 5, 15, 30, 60, 120, 300]
)
token_usage = Counter(
    "agent_tokens_total",
    "Token 使用总量",
    ["model", "token_type"]    # token_type: input/output
)
active_agents = Gauge(
    "agent_active_runs",
    "当前正在运行的 Agent 数量"
)

class AgentMonitor:
    def __enter__(self):
        self.start_time = time.time()
        active_agents.inc()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        agent_duration.observe(duration)
        active_agents.dec()
        status = "failed" if exc_type else "success"
        agent_runs_total.labels(status=status, agent_type="research").inc()

# 使用方式
async def monitored_agent_run(user_input: str) -> str:
    with AgentMonitor():
        result = await run_agent_with_timeout(user_input)
        return result

# 启动 Prometheus metrics 端口（Grafana 读取）
# start_http_server(8000)

部署架构推荐