AI Agent开发与安全

AI Agent的核心演进

1. Level 1：LLM Agent（聊天机器人）

   • 特点：靠提示词工程赋予人设（如星座占卜、角色扮演），但存在幻觉问题，输出不可控。
   • 局限：娱乐性强，难胜任严肃任务。

2. Level 2：AI Agent（实用工具）

   • 升级点：新增 规划（拆解任务步骤）、记忆（存储上下文）、工具使用（调用API/数据库）。
   • 代表：OpenManus、OWL等，能处理复杂任务（如自动化办公）。

3. Level 3：Multi-Agent（多智能体协作）

   • 为什么需要？ 单Agent难精通所有领域（如同时懂编程+画图）。
   • 方案：多个专业Agent分工协作（如产品经理Agent分析需求 + 程序员Agent写代码）。
   • 关键优势：
     • 任务拆解更高效
     • 可独立优化单个Agent
     • 支持人机协同（Human in the Loop）

两大核心协议：MCP 与 A2A

1. MCP（面向工具调用）

• 目标：让AI统一调用外部工具（如天气API、数据库）。
• 架构：
  • 主机（如IDE插件）→ 客户端（连接器）→ 服务器（工具实现）。
• 价值：工具一次开发，多模型通用（类似USB-C接口）。

2. A2A（面向Agent协作）

• 目标：让多个Agent互相发现、分工协作（如旅行规划Agent + 酒店预订Agent）。
• 核心功能：
  • 发布能力（Agent Card）
  • 任务分发与状态同步
  • 结果流式返回
• 优势：打破厂商壁垒，构建开放生态。

3. MCP vs A2A 关系

• 互补：MCP管工具调用，A2A管Agent协作。
• 竞争可能：高级工具 ≈ 弱Agent，未来协议可能融合。

Agent的思考框架

1. 思维链（CoT）

   • 让模型分步骤推理（如先分析问题再生成答案），提升逻辑性。

2. ReAct（推理+行动）

   • 循环流程：思考 → 行动（调用工具）→ 观察结果 → 再思考。
   • 适用：需动态调整的任务（如实时搜索+分析）。

3. Plan-and-Execute（先规划后执行）

   • 步骤：
     1. 生成完整计划（如“1. 搜索资料 → 2. 总结 → 3. 生成报告”）。
     2. 严格按步骤执行，避免中途被干扰。
   • 优势：适合长链条任务，稳定性强。

Golang开发框架：Eino

1. 核心设计

• 强类型：用Go泛型确保节点输入输出类型安全。
• 组件化：预置ChatModel、工具调用等模块，开箱即用。
• 编排能力：用有向图连接组件，可视化构建流程（类似流程图）。

2. 关键功能

• Callback机制：嵌入日志、监控等非业务逻辑（如追踪工具调用耗时）。
• Checkpoint：支持人机协同（如任务中途暂停让用户确认）。
• 示例代码：10行实现天气查询Agent（结合MCP调用高德地图API）。

多Agent系统落地实践

1. 架构设计（Supervisor模式）

   • 意图识别Agent（客户经理） → 分发任务给领域Agent（如旅行规划、深度搜索）。
   • 所有Agent通过A2A协议互联。

2. 连接生态

   • Cherry Studio：通过OpenAI兼容接口快速接入Agent。
   • QQ机器人：用A2A协议将Agent能力植入QQ生态。

3. 可观测性

   • 集成 Langfuse：实时查看任务链路、性能指标、错误日志。
   • 10行代码接入，无侵入式监控。

AI Agent开发心法

1. 协议是基石：MCP统一工具调用，A2A实现多Agent协作。
2. 框架提效率：Eino解决Go生态工程化问题。
3. 设计模式是关键：
   • 简单任务 → ReAct（边想边做）
   • 复杂任务 → Plan-and-Execute（先规划后执行）
4. 生态扩展：通过Connector快速接入QQ、Cherry Studio等平台。

结论：
AI Agent正从“玩具”走向“工具”，多Agent协作+标准化协议+工程化框架是高效开发的核心。选择适合场景的协议（MCP/A2A）和思考框架，结合Eino等工具，即可优雅构建复杂Agent系统。

技术选型对比表

抓住协议、框架、设计模式三条主线，就能理解复杂Agent系统的全貌。

架构核心组件详解

1. LLM（大语言模型）

• 核心作用：自然语言理解与任务规划
• 关键技术：
  • 思维链（CoT）：将用户指令分解为可执行步骤

    # 示例：LLM任务分解
    prompt = f"""
    用户请求：{user_query}
    可用工具：{tool_list}
    输出JSON格式：{"steps": [{"tool": "tool_name", "params": {...}}]}
    """
    

  • 约束生成：限定输出格式防止幻觉

2. Agent（智能代理）

• 核心引擎：任务调度与决策
• 状态机实现：

  class Agent:
      def __init__(self, tools):
          self.state = "IDLE"
          self.working_memory = []  # 存储中间结果
      
      def execute_step(self, step):
          # 工具选择算法（基于余弦相似度）
          tool = max(self.tools, key=lambda t: cosine_sim(step["desc"], t.desc))
          result = tool.execute(step["params"])
          self.working_memory.append(result)
          
          # 状态转移
          if "需要进一步处理" in result:
              self.state = "NEED_FOLLOWUP"
          else:
              self.state = "AWAITING_NEXT"
  

3. MCP（模型上下文协议）

• 协议本质：统一工具调用规范
• 协议结构：

  {
    "version": "1.0",
    "action": "EXECUTE",
    "payload": {
      "tool_id": "crm_sales_query",
      "parameters": {"time_range": "last_quarter"},
      "auth_context": {"user_id": "U123", "token": "xyz"}
    }
  }
  

• 安全设计：
  • 参数白名单：限制可传递的敏感字段
  • 执行沙箱：危险操作隔离运行

    def safe_execute(tool, params):
        with Sandbox() as sandbox:
            return sandbox.run(tool.executable, params)
    

全链路工作流程（14步详解）

阶段1：请求解析（1-3步）

1. 用户输入：自然语言请求（例：“统计Q3华北区销售额”）
2. LLM解析：
   • 生成结构化指令：

     {
       "steps": [
         {"tool": "region_filter", "params": {"region": "华北"}},
         {"tool": "sales_aggregator", "params": {"period": "2023Q3"}}
       ]
     }
     

3. 权限预检：验证用户是否有权调用指定工具

阶段2：安全拦截（4-5步）

4. 操作确认：向用户展示即将执行的操作

   [系统提示] 将执行： 
   1. 在CRM系统筛选"华北"区域数据 
   2. 汇总2023年第三季度销售额
   确认执行？(Y/N)
   

5. 用户授权：获得明确授权后才继续

阶段3：协议执行（6-11步）

6. 协议封装：Agent生成MCP请求

   mcp_request = {
     "tool_id": "crm_api",
     "parameters": {"action": "get_sales", "filters": {...}}
   }
   

7. 服务路由：MCP Client通过服务发现定位目标系统
8. 协议转换：将MCP请求转换为目标API格式

   # 转换示例（CRM系统适配器）
   def adapt_to_crm(mcp_req):
       return {
         "endpoint": "/v1/sales/query",
         "body": {"filter": mcp_req["parameters"]["filters"]}
       }
   

9. 执行监控：记录请求耗时、成功率等指标
10. 异常处理：
    • 重试机制：对超时请求自动重试2次
    • 熔断保护：失败率超阈值时暂停调用

阶段4：结果生成（12-14步）

11. 数据增强：LLM对原始数据加工

    # 示例：生成可视化建议
    if result["data_type"] == "sales_report":
        return llm_generate("将数据转为折线图代码")
    

12. 审计日志：记录完整操作轨迹

    [审计日志] 用户U123 执行CRM查询 → 结果大小1024行 → 生成图表
    

13. 交付输出：最终结果返回用户界面

企业级关键技术实现

1. 混合知识引擎

• 三路数据融合：

  def query_knowledge(question):
      # 并行查询
      vector_result = vector_db.semantic_search(question) 
      graph_result = neo4j.query(f"MATCH (n)-[r]->(m) WHERE n.name CONTAINS '{question}' RETURN n,r,m")
      sql_result = db.execute(llm_to_sql(question))
      
      # 结果融合算法
      return FusionEngine.merge(
        sources=[vector_result, graph_result, sql_result],
        weights=[0.4, 0.3, 0.3]  # 可配置权重
      )
  

2. 动态Agent编排

• 工作流引擎：

  class WorkflowEngine:
      def __init__(self, dag):
          self.dag = dag  # 有向无环图
      
      def run(self, inputs):
          for node in topological_sort(self.dag):
              if node.type == "TOOL":
                  outputs = execute_tool(node.tool_id, inputs)
              elif node.type == "LLM":
                  outputs = llm_process(node.prompt, inputs)
              inputs = outputs  # 传递中间结果
          return outputs
  

3. 安全控制体系

生产环境部署建议

1. 性能优化

• LLM缓存：对重复查询缓存思维链结果

  @cache(ttl=300, key_fn=lambda q: hash(q))
  def get_llm_plan(query):
      return llm.generate(prompt_template.format(query))
  

• 连接池：MCP Client维护数据库/API连接池

2. 高可用设计

• Agent集群：ZK选主实现故障转移
• 流量调度：

  upstream agent_nodes {
    server 10.0.0.1:8000 weight=5; 
    server 10.0.0.2:8000;
    check interval=3000 rise=2 fall=3;
  }
  

3. 演进机制

• 持续训练：

  def online_learning(feedback):
      # 收集负样本
      if feedback.rating < 3:
          save_training_data(feedback.query, feedback.expected)
      # 每周微调
      if time.now().weekday() == 0:
          retrain_model(training_data)
  

典型问题解决方案

问题：工具调用冲突

• 解决方案：乐观锁+事务补偿

  def execute_transaction(steps):
      try:
          with transaction():
              for step in steps:
                  tool.lock(resource_id)
                  result = tool.execute(step)
                  tool.unlock(resource_id)
      except ConflictError:
          # 补偿机制
          for step in reversed(completed_steps):
              tool.compensate(step)
  

问题：长流程中断

• 解决方案：状态持久化

  class StateManager:
      def save_state(self, agent):
          redis.set(f"agent:{agent.id}", 
                    pickle.dumps(agent.working_memory))
      
      def recover_state(self, agent_id):
          return pickle.loads(redis.get(f"agent:{agent_id}"))
  

MCP+LLM+Agent架构本质是将自然语言编译为系统API调用的技术栈，其核心价值在于：

1. 统一协议层：通过MCP消除N×M集成复杂度
2. 动态编排器：Agent实现跨系统工作流自动化
3. 持续进化力：在线学习机制使系统越用越智能

企业落地建议：从客服工单处理、IT运维等标准化场景切入，逐步构建企业级AI Agent中台。参考技术栈：LangChain（Agent框架）+ FastAPI（MCP服务）+ Weaviate（向量库）。

核心问题：Prompt注入攻击

• 是什么？ 攻击者把恶意指令藏在看起来正常的输入（如用户提问、网页内容、上传文件）里，骗AI去执行不该做的事情（比如删除文件、泄露数据、发垃圾邮件）。
• 为什么头疼？ AI被训练成“听话”的助手，很难分辨哪些是用户真需求，哪些是藏起来的坏指令。特别是当AI有权限操作外部工具（执行代码、访问网络、管理设备）时，风险更大。
• 目标： 让AI做坏事，或者让它“罢工”。

核心结论：没有绝对安全，但可以大大加强防御！

想做一个对所有Prompt注入都免疫的AI几乎不可能（因为它总要理解文字）。更实际的办法是改变AI系统的设计架构，限制它的能力，即使被“忽悠”了，也干不成坏事或坏事影响有限。

六种“安全设计模式”详解（就像给AI建安全屋）：

1. 行动选择器模式：

   • 核心： 一次只做一件事，不看结果反馈。
   • 怎么防注入： AI收到指令（可能含恶意指令），决定下一步行动（比如“搜索天气”）。系统执行这个行动（比如显示搜索结果），但行动结果不给AI看！ AI决定下一个行动时，只看最初指令和内部状态，不看那些可能藏着坏指令的结果。
   • 比喻： 指挥官（AI）下命令“侦察A区”，侦察兵（系统）去了，回来报告。但指挥官不看报告内容，只根据最初任务决定下个命令是“进攻”还是“撤退”。报告里就算写着“叛变吧！”，指挥官也看不见。
   • 适用： 简单任务，不需要根据结果做复杂决策。

2. 计划-执行模式：

   • 核心： 先列计划表，然后死板执行。
   • 怎么防注入： AI先根据任务，制定一个完整的、多步骤的计划（比如：1. 搜索科技新闻；2. 阅读结果；3. 写总结）。计划一旦定好，系统就严格按照计划步骤执行。执行过程中产生的数据（如网页内容）可能含坏指令，但不会送回AI去修改计划。
   • 比喻： 指挥官（AI）战前制定好作战计划书（搜索->阅读->总结）。士兵（系统）拿到计划书就严格执行。打到一半发现敌人阵地上有标语“投降吧！”，但士兵只按计划书行动，不会因此改变计划去投降。
   • 适用： 多步骤任务，但任务流程相对固定，不需要中途灵活调整计划。

3. LLM Map-Reduce模式：

   • 核心： 分工协作，让“危险”AI干安全的活儿。
   • 怎么防注入： 系统有两个AI：
     • 协调者（主AI）： 负责分解任务、分派工作、汇总结果。它不接触原始危险输入。
     • 工作者（子AI）： 负责处理可能含坏指令的输入（如用户文件），但能力被严格限制，只能做特定安全操作（如按固定格式提取信息）。它们无法执行坏指令要求的危险操作。
   • 比喻： 老板（协调者）让一群实习生（工作者）从一堆匿名信（危险输入）里只抄写信封上的地址（安全任务）。实习生只能抄地址，无法执行信里写的“烧掉办公室”这种指令。老板只收集整理实习生抄好的地址（安全结果）。
   • 适用： 处理大量不可信内容（如用户上传文件、评论）进行信息提取、分类等安全操作。

4. 双重LLM模式：

   • 核心： “翻译官”+“执行官”，危险输入不直接给“执行官”。
   • 怎么防注入： 系统有两个AI：
     • 隔离LLM： 负责处理原始用户输入/外部内容（可能含坏指令）。它能力受限，不直接执行操作，只把输入翻译成安全符号（如 INTENT_开灯, LOCATION_客厅）。
     • 特权LLM： 只接收隔离LLM翻译好的安全符号，并据此决定和执行操作（如开灯）。它从不接触危险的原始输入。
   • 比喻： 你对外交官（特权LLM）说：“客厅好暗，开灯吧”（可能被窃听成“发射导弹”）。你的翻译（隔离LLM）只告诉外交官：“用户想开客厅灯”（安全符号）。外交官根据这个开灯。窃听者喊的“发射导弹！”不会被翻译成有效指令。
   • 适用： 需要理解用户意图并执行敏感操作（如智能家居、账户管理），尤其是语音、自然语言交互场景。

5. 代码生成-执行模式：

   • 核心： AI写“安全剧本”，“演员”在“安全屋”表演。
   • 怎么防注入： 一个受保护的AI（主智能体）根据任务，编写一段“安全代码”（用特定、受限的语言写成）。这段代码定义了操作流程。代码在一个隔离的沙箱环境中执行。主AI不执行用户指令，只负责生成代码和接收执行后的结构化结果。
   • 比喻： 编剧（主AI）写了个剧本（安全代码），规定演员（沙箱）只能做“拿起A道具”、“念B台词”等安全动作。演员在封闭舞台（沙箱）严格按剧本表演。用户即使喊“把舞台炸了！”，剧本里没这动作，演员就不会做。
   • 适用： 自动化流程、数据处理等需要执行多个工具操作的任务，对安全性要求极高。

6. 上下文最小化模式：

   • 核心： 给AI“喂饭”要少而精，别啥都给它看。
   • 怎么防注入： 在把信息（对话历史、文档、搜索结果）交给AI做决策参考前，先过滤、裁剪或总结，只留下当前任务绝对必需的信息，删除无关或可能有害的部分。
   • 比喻： 不让厨师（AI）直接进乱糟糟的仓库（原始上下文），而是由助手先按订单（当前任务）挑出需要的几种食材（最小化上下文），洗干净切好再交给厨师。仓库里藏的“毒蘑菇”（恶意指令）很可能就被助手剔除了。
   • 适用： 通用策略，可与其他模式结合使用。尤其适用于基于长对话历史或多文档进行响应的场景。

如何选择？

• 没有万能药： 每种模式各有优缺点和适用场景，需要根据你的AI具体做什么、安全要求多高来选择。
• 核心思想一致：
  • 隔离危险： 不让AI直接接触可能藏有恶意指令的原始文本。
  • 限制能力： 让处理危险输入的组件只能做安全的、受限的操作。
  • 结构化交互： 使用代码、符号、固定格式等结构化方式传递意图和结果，避免依赖自由文本解析。
  • 控制信息流： 精心设计哪些信息在哪个阶段被哪个组件看到。
• 组合使用： 常常需要组合使用多种模式来构建更健壮的系统（例如：上下文最小化 + 计划-执行模式 + 代码生成-执行模式）。

把这六种模式想象成给AI系统设计的安全流程和权限管理。核心就是：别让AI什么都看、什么都信、什么都能做！ 通过设计好的“流水线”、“防火墙”和“操作手册”，让即使部分环节被“忽悠”了，整个系统也不会出大乱子。选择哪种“安全屋”设计，取决于你要让AI干什么活儿。