AI Agents：掌握3个关键模式（Reflexive）。第3部分，共3部分

Source: Dev.to

这些模式的代码已在 GitHub 上提供。
仓库

反射模式 – 当没有时间思考时

在 Article 2 (ReAct) 中，我们把代理人变成了细致的 “调查员”。
代理人会坐下来，分析、计划、进行 Google 搜索，然后再作答。
这对复杂任务非常有效，但在现实世界中它有一个巨大的 缺陷：速度慢。

可以这样想：如果你碰到热炉子，你的大脑不会开始辩论：

“嗯，我检测到温度是 200 °C，这会导致蛋白质变性，我建议移动肱二头肌……”

哈哈，不会的。你的脊髓会立刻发号施令，在你甚至感受到疼痛之前就把手抽回来。

反射模式（有时称为 语义路由器）正是如此：人工智能的 “反射动作”。

Source: …

用例：安全哨兵

想象一个系统能够 实时 读取你的服务器日志。在成千上万的正常访问中，出现了一条恶意请求：

GET /search?q=' OR 1=1;--

如果在这里使用 ReAct 代理，你就完蛋了。代理会这样想：

“哇，这是一种不寻常的模式。我会使用搜索工具来了解‘SQL 注入’是什么。啊，我看到它很危险。准备阻断。”

等它“思考”完（10‑15 秒）时，你的数据库已经在暗网泄露。

Reflexive Agent 的优势所在

示例反射动作（Python）

def block_ip(ip: str, reason: str) -> str:
    """立即阻断恶意 IP 地址。"""
    msg = f"⛔ BLOCKING IP [{ip}]: {reason}"
    logger.warning(msg)
    return msg

def escalate_to_admin(log_id: str, severity: str) -> str:
    """将关键问题上报给人工管理员。"""
    msg = f"⚠️ ADMIN ALERT: Log [{log_id}] - Severity [{severity}]"
    logger.critical(msg)
    return msg

def log_normal_traffic(source: str) -> str:
    """记录合法且安全的流量。"""
    msg = f"✅ Normal traffic from [{source}]"
    logger.info(msg)
    return msg

# ----------------------------------------------------------------------
# 4. Agno Agent Configuration (Reflexive Pattern)
# ----------------------------------------------------------------------
model_id = os.getenv("BASE_MODEL", "gpt-4o-mini")

agent = Agent(
    model=OpenAIChat(id=model_id),
    tools=[block_ip, escalate_to_admin, log_normal_traffic],
    instructions=[
        "You are an automated security system (Reflexive Pattern).",
        "Your job is to classify incoming logs and execute the corresponding tool IMMEDIATELY.",
        "Action rules:",
        "- If you detect SQL Injection, XSS, or known attacks → use `block_ip`.",
        "- If you detect server errors (500), service downs, or timeouts → use `escalate_to_admin`.",
        "- If traffic seems legitimate, successful login, or normal navigation → use `log_normal_traffic`.",
        "DO NOT explain your decision. DO NOT greet. Just execute the tool directly in response to the stimulus.",
        "Process the log and terminate execution."
    ]
)

# ----------------------------------------------------------------------
# 5. Flow Simulation (The Stream)
# ----------------------------------------------------------------------
def main():
    logger.info("Starting Security Sentinel Agent (Reflexive)...")
    print("--- Security Sentinel - Reflexive Pattern ---")

    # Simulated logs
    logs = [
        "User 'admin' logged in successfully from 192.168.1.10",
        "GET /search?q=' OR 1=1;-- (SQL Injection attempt from 203.0.113.5)",
        "Service 'PaymentGateway' timed out after 3000ms (Error 500)",
        "User 'jdoe' updated profile picture",
        "alert('hacked') sent via form from 88.22.11.44",
        "DB_ERROR: Connection pool exhausted (LogID: 4452)",
        "Bot crawler identified: Googlebot/2.1 from 66.249.66.1",
        "Multiple failed login attempts for user 'sys'..."
        # ... additional logs could be added here
    ]

    for entry in logs:
        # The agent receives each log entry as a stimulus and reacts instantly.
        response = agent.run(entry)
        print(response)

if __name__ == "__main__":
    main()

上述代码演示了 纯粹的反射式 工作流：代理收到一行日志后，根据硬编码规则匹配并立即触发相应工具——没有思考、没有循环，亚秒级响应。

def main():
    logs = [
        "admin' from 45.33.22.11",
        "HealthCheck: All systems operational - latency 12ms",
        "Unauthorized access attempt to /etc/passwd from 172.16.0.50"
    ]

    print(f"\nProcessing {len(lo")

gs)} logs in the 'stream'...\n")

    for i, log_entry in enumerate(logs, 1):
        try:
            print(f"[{i}] Stimulus: {log_entry}")
            logger.info(f"Processing log {i}: {log_entry}")

            # In reflexive mode, we expect the agent to execute the tool immediately.
            # We use show_tool_calls=True to demonstrate the action.
            agent.print_response(log_entry, show_tool_calls=True)
            print("-" * 50)

        except Exception as e:
            logger.error(f"Error processing log {i}: {str(e)}")
            print(f"An error occurred in event {i}: {e}")

if __name__ == "__main__":
    main()

限制性指令（负约束）

在系统提示（instructions）中，我们明确禁止代理进行思考或聊天：

• 不要解释
• 只执行

代理仅被编程为触发工具。

直接映射

没有反馈回路。流程是线性的：

Incoming Log → Classification → Tool Execution

模型选择

• 已配置 gpt-4o-mini。
• 使用更小、更快的模型以优先考虑延迟和处理量。

为什么这个模式很酷（优点）

• 荒诞的速度（极低延迟）： 由于不生成“思考”文本，响应几乎是瞬时的。
• 高吞吐量： 非常适合每分钟处理数千封电子邮件、实时聊天或监控警报。
• 可预测（确定性）： 去除“创造力”使代理行为像经典脚本——给定 X，它总是执行 Y。
• 成本效益高： 更少的输出 token 和更小的模型 = 更低的 API 费用。

失效之处（缺点）

• 金鱼记忆（情境盲点）: 代理仅对当前输入作出反应。如果一次攻击被分成三步，跨越 10 分钟进行，它不会将这些步骤联系起来。
• 顽固（僵硬）: 对于其规则未覆盖的模糊输入，可能导致失败或错误的分类。
• 黑箱: 与 ReAct 不同，缺少“思考”日志来解释为何采取了某个动作。

开发者对比

来自实战的技巧

1. 当心“封禁锤”

由于该代理先开火后提问，误报很常见。
建议： 除非你 100 % 确定，否则将操作设为 flag_for_review() 而不是 delete_account()。

2. 对提示词要严格（负向约束）

• 错误提示词： “对这段文字进行分类。”
• 正确提示词： “对这段文字进行分类。不要解释原因。不要生成引言文本。只返回 JSON。”

3. 混合架构

反射模式是 “步兵”。 将其部署在前线，过滤约 80 % 的简单噪声。
如果它遇到不理解或模糊的情况，升级工单给 ReAct 代理（更聪明但更昂贵）进行深入调查。这样可以兼顾两者的优势。

祝编码愉快！ 🤖