编程 AI Agent 框架实战:从0到1构建智能代理系统(2026实战指南)

2026-07-05 03:47:01 +0800 CST views 10

AI Agent 框架实战:从0到1构建智能代理系统(2026实战指南)

本文深入剖析 AI Agent 的核心架构、主流框架对比、实战代码演练,以及生产环境最佳实践。无论你是刚接触 Agent 开发,还是准备将 Agent 投入生产,这篇文章都能给你带来实实在在的收获。

目录

  1. AI Agent 是什么?
  2. 核心概念:ReAct 与 Plan-and-Execute
  3. 主流框架对比
  4. 架构设计:生产级 Agent 系统
  5. 代码实战:实现一个多功能 Agent
  6. 工具调用深度剖析
  7. 记忆系统与上下文管理
  8. 性能优化
  9. 安全与合规
  10. 部署与监控

1. AI Agent 是什么?

1.1 从 ChatGPT 到 Agent:范式的跃迁

2022年11月,ChatGPT 发布,全世界看到了大语言模型(LLM)的威力。但 ChatGPT 本质上是一个对话系统——你问,它答,仅此而已。

2026年的今天,AI 已经从"能说会道"进化到"能做事"。这就是 AI Agent(智能代理) 的核心价值:

Agent = LLM + 工具调用 + 记忆 + 规划能力

一个简单的对比:

特性传统对话模型AI Agent
交互方式问答任务执行
工具使用
记忆能力短期上下文长期记忆
规划能力
自主性被动响应主动规划

1.2 Agent 的核心特征

一个真正的 AI Agent 应该具备以下特征:

class TrueAgent:
    """
    真正的人工智能代理
    
    特征:
    1. 感知(Perception):理解用户输入和环境状态
    2. 规划(Planning):将复杂任务分解为子任务
    3. 行动(Action):调用工具执行具体操作
    4. 记忆(Memory):维持长期上下文
    """
    
    def __init__(self, llm, tools, memory):
        self.llm = llm           # 语言模型
        self.tools = tools       # 可用工具集
        self.memory = memory     # 记忆系统
    
    def run(self, task: str) -> str:
        """执行任务的主循环"""
        # 1. 感知:理解任务
        context = self.perceive(task)
        
        # 2. 规划 + 执行循环
        for i in range(self.max_iterations):
            # 规划下一步行动
            plan = self.plan(context)
            
            # 如果任务完成,退出
            if plan.is_complete:
                return plan.result
            
            # 执行行动
            action_result = self.act(plan.next_action)
            
            # 更新上下文
            context = self.update_context(context, action_result)
        
        return "任务未能在限定步骤内完成"

2. 核心概念:ReAct 与 Plan-and-Execute

2.1 ReAct:最经典的 Agent 范式

ReAct(Reasoning + Acting) 是2022年普林斯顿大学提出的 Agent 范式,至今仍是最广泛使用的模式。

核心思想:让 LLM 在每一步都进行推理(Reasoning),然后行动(Acting),观察结果,再推理...

ReAct 的 Prompt 模板

任务:{task}

你可以使用以下工具:
{tools}

请按照以下步骤思考并行动:
Thought: [你现在想什么?]
Action: [工具名称],输入:[工具输入]
Observation: [工具返回的结果]
...(重复以上步骤)
Thought: 我现在知道最终答案了
Final Answer: [最终答案]

代码示例:一个简单的 ReAct Agent

from typing import List, Dict, Any
import re

class ReActAgent:
    """ReAct 范式的 Agent 实现"""
    
    def __init__(self, llm, tools: List[Dict]):
        self.llm = llm
        self.tools = {tool["name"]: tool for tool in tools}
        
    def parse_action(self, text: str) -> Dict[str, Any]:
        """解析 LLM 输出的 Action"""
        pattern = r"Action:\s*(\w+),输入:\s*(.+)"
        match = re.search(pattern, text)
        if match:
            return {
                "tool": match.group(1),
                "input": match.group(2)
            }
        return None
    
    def execute_tool(self, tool_name: str, tool_input: str) -> str:
        """执行工具调用"""
        if tool_name not in self.tools:
            return f"错误:工具 {tool_name} 不存在"
        
        tool = self.tools[tool_name]
        try:
            result = tool["func"](tool_input)
            return result
        except Exception as e:\n            return f"工具执行失败:{str(e)}"
    
    def run(self, task: str, max_steps: int = 10) -> str:
        """运行 ReAct Agent"""
        
        # 构建初始 prompt
        prompt = f"""
任务:{task}

可用工具:{', '.join(self.tools.keys())}

请按照 ReAct 范式思考并行动:
Thought: [你的思考过程]
Action: [工具名],输入:[参数]
Observation: [观察结果]
...(重复)
Thought: 我现在知道最终答案了
Final Answer: [最终答案]
"""
        
        conversation = prompt
        
        for step in range(max_steps):
            # 调用 LLM
            response = self.llm(conversation)
            conversation += "\n" + response
            
            # 解析 Action
            action = self.parse_action(response)
            
            if action is None:
                # 检查是否完成了
                if "Final Answer:" in response:
                    final_answer = response.split("Final Answer:")[1].strip()
                    return final_answer
                continue
            
            # 执行工具
            observation = self.execute_tool(action["tool"], action["input"])
            
            # 将观察结果加回对话
            conversation += f"\nObservation: {observation}\nThought:"
        
        return "达到最大步骤限制,任务未完成"

# 使用示例
def search_tool(query: str) -> str:
    """模拟搜索工具"""
    return f"搜索结果:关于 '{query}' 的详细信息..."

def calculator_tool(expr: str) -> str:
    """计算器工具"""
    try:
        result = eval(expr)
        return f"计算结果:{result}"
    except:
        return "计算失败"

tools = [
    {"name": "Search", "func": search_tool},
    {"name": "Calculator", "func": calculator_tool}
]

# 运行 Agent
agent = ReActAgent(llm=your_llm, tools=tools)
result = agent.run("请计算 234 * 567,然后搜索这个结果的相关信息")
print(result)

2.2 Plan-and-Execute:更适合复杂任务

ReAct 适合短链条任务,但对于需要多步规划的任务,容易出现"短视"问题。

Plan-and-Execute 范式将过程分为两个阶段:

  1. 规划阶段(Plan):先制定完整计划
  2. 执行阶段(Execute):按计划逐步执行

优势

  • 全局视野,避免陷入局部最优
  • 计划可以人工审核,提高可控性
  • 适合复杂、多步骤任务

3. 主流框架对比

3.1 LangChain:生态最丰富的 Agent 框架

GitHub Star: 95k+
适用场景: 快速原型、复杂工作流

优势

  • 生态最丰富,集成最多(LLM、向量库、工具)
  • 灵活的链式(Chain)和 Agent 抽象
  • 大量教程和社区支持

劣势

  • 抽象层次过多,学习曲线陡
  • 性能开销大(大量中间层)
  • 版本迭代快,API 不稳定

核心概念

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI

# 1. 定义工具
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气:晴,25°C"

tools = [
    Tool(
        name="Weather",
        func=get_weather,
        description="获取城市天气。输入:城市名称"
    )
]

# 2. 创建 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 3. 创建 Agent
agent = create_openai_functions_agent(llm, tools, prompt)

# 4. 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5
)

# 5. 运行
result = agent_executor.invoke({"input": "北京今天天气怎么样?"})
print(result["output"])

3.2 AutoGPT:自主性的巅峰

GitHub Star: 172k+
适用场景: 完全自主任务、研究实验

核心特点

  • 完全自主:给定目标后,Agent 自己规划、执行、反思
  • 长期记忆:使用向量数据库存储记忆
  • 联网能力:自动搜索、浏览网页

AutoGPT 的问题

  • 太自主:难以控制,经常"跑偏"
  • 成本高:大量 LLM 调用,API 费用昂贵
  • 不稳定:容易陷入循环或产生错误行动

适用建议:适合研究,生产环境慎用。

3.3 CrewAI:多 Agent 协作的新秀

GitHub Star: 20k+
适用场景: 多角色协作、复杂工作流

核心概念

  • Crew(团队): 多个 Agent 组成团队
  • Agent(角色): 每个 Agent 有特定角色和工具
  • Task(任务): 分配给 Agent 的具体任务
  • Process(流程): 顺序、并行、层次化

代码示例

from crewai import Agent, Task, Crew, Process

# 1. 定义 Agent(角色)
researcher = Agent(
    role='研究员',
    goal='深入研究给定话题',
    backstory="你是一位经验丰富的研究员,擅长信息收集和分析",
    tools=[search_tool, web_scraper_tool],
    verbose=True
)

writer = Agent(
    role='技术作家',
    goal='将研究结果转化为高质量文章',
    backstory="你是一位资深技术作家,擅长将复杂技术讲清楚",
    tools=[markdown_tool],
    verbose=True
)

# 2. 定义 Task(任务)
research_task = Task(
    description='研究 "AI Agent 框架" 的最新进展',
    expected_output='一份详细的研究报告',
    agent=researcher
)

write_task = Task(
    description='基于研究报告,撰写一篇3000字的深度文章',
    expected_output='一篇结构清晰、有深度的技术文章',
    agent=writer
)

# 3. 创建 Crew(团队)
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True
)

# 4. 启动
result = crew.kickoff()
print(result)

3.4 框架选择建议

需求推荐框架理由
快速原型LangChain生态丰富,上手快
完全自主AutoGPT自主性最强
多角色协作CrewAI原生支持多 Agent
生产环境自写 + LiteLLM可控性、性能

4. 架构设计:生产级 Agent 系统

4.1 生产级 Agent 的挑战

开发环境的 Demo 和生产环境的要求截然不同:

维度Demo生产
可靠性能跑就行99.9% 可用
成本不考虑必须优化
安全无要求严格管控
监控print 就行完整可观测性

4.2 生产级 Agent 架构

一个生产级 Agent 系统应该包含以下模块:

┌─────────────────────────────────────────────────┐
│                   API Gateway                    │
│              (认证、限流、路由)                    │
└─────────────────┬───────────────────────────────┘
                  │
┌─────────────────▼───────────────────────────────┐
│                Agent Orchestrator                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ 任务解析  │  │ 规划引擎  │  │ 执行引擎  │      │
│  └──────────┘  └──────────┘  └──────────┘      │
└─────────────────┬───────────────────────────────┘
                  │
      ┌───────────┼───────────┐
      │           │           │
┌─────▼─────┐ ┌──▼────┐ ┌───▼────┐
│ 工具注册表  │ │ 记忆系统 │ │  LLM   │
│ (Registry) │ │(Memory)│ │  Pool  │
└────────────┘ └───────┘ └────────┘

4.3 核心模块详解

4.3.1 工具注册表(Tool Registry)

from dataclasses import dataclass
from typing import Callable, Dict, List

@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    func: Callable
    parameters: Dict[str, str]
    required_permissions: List[str]
    timeout: int = 30

class ToolRegistry:
    """工具注册表"""
    
    def __init__(self):
        self._tools: Dict[str, Tool] = {}
    
    def register(self, tool: Tool):
        """注册工具"""
        if tool.name in self._tools:
            raise ValueError(f"工具 {tool.name} 已存在")
        self._tools[tool.name] = tool
    
    def get(self, name: str) -> Tool:
        """获取工具"""
        return self._tools.get(name)
    
    def validate_permissions(self, tool_name: str, user_permissions: List[str]) -> bool:
        """验证用户是否有权限使用工具"""
        tool = self._tools.get(tool_name)
        if not tool:
            return False
        return all(perm in user_permissions for perm in tool.required_permissions)

4.3.2 记忆系统(Memory System)

class MemorySystem:
    """分层记忆系统"""
    
    def __init__(self, vector_db, llm):
        self.vector_db = vector_db
        self.llm = llm
        self.short_term = []
        self.max_short_term = 10
    
    def add(self, content: str, metadata: Dict = None):
        """添加记忆"""
        self.short_term.append({
            "content": content,
            "timestamp": time.time(),
            "metadata": metadata or {}
        })
        
        if len(self.short_term) > self.max_short_term:
            old_memory = self.short_term.pop(0)
            self.vector_db.add(
                text=old_memory["content"],
                metadata=old_memory["metadata"]
            )
    
    def retrieve(self, query: str, top_k: int = 5) -> List[str]:
        """检索相关记忆"""
        # 1. 从短期记忆中检索
        recent = [m["content"] for m in self.short_term[-5:]]
        
        # 2. 从长期记忆中检索(语义搜索)
        long_term = self.vector_db.search(query, top_k=top_k)
        
        # 3. 合并并去重
        all_memories = recent + [m["text"] for m in long_term]
        return list(dict.fromkeys(all_memories))

5. 代码实战:实现一个多功能 Agent

5.1 完整代码实现

"""
生产级多功能 Agent 实现
功能:工具调用、记忆管理、任务规划、错误处理
"""

import json
import time
import logging
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from abc import ABC, abstractmethod
import re

# ==================== 数据结构定义 ====================

@dataclass
class ToolResult:
    """工具执行结果"""
    success: bool
    output: str
    error: Optional[str] = None
    execution_time: float = 0.0

# ==================== 工具系统 ====================

class BaseTool(ABC):
    """工具基类"""
    
    @abstractmethod
    def name(self) -> str:
        """工具名称"""
        pass
    
    @abstractmethod
    def description(self) -> str:
        """工具描述"""
        pass
    
    @abstractmethod
    def execute(self, **kwargs) -> ToolResult:
        """执行工具"""
        pass

class SearchTool(BaseTool):
    """搜索工具(模拟)"""
    
    def name(self) -> str:
        return "search"
    
    def description(self) -> str:
        return "网络搜索工具。输入:搜索关键词。"
    
    def execute(self, query: str, **kwargs) -> ToolResult:
        start_time = time.time()
        
        try:
            # 模拟搜索 API 调用
            time.sleep(0.5)
            
            mock_results = {
                "Python": "Python 是一种高级编程语言",
                "AI Agent": "AI Agent 是能够自主规划的智能系统"
            }
            
            result = mock_results.get(query, f"关于 '{query}' 的搜索结果")
            
            return ToolResult(
                success=True,
                output=result,
                execution_time=time.time() - start_time
            )
        
        except Exception as e:\n            return ToolResult(\n                success=False,\n                output="",
                error=str(e),
                execution_time=time.time() - start_time
            )

class CalculatorTool(BaseTool):
    """计算器工具"""
    
    def name(self) -> str:
        return "calculator"
    
    def description(self) -> str:
        return "数学计算器。输入:数学表达式。"
    
    def execute(self, expression: str, **kwargs) -> ToolResult:
        start_time = time.time()
        
        try:
            allowed_chars = set('0123456789+-*/.() ')
            if not all(c in allowed_chars for c in expression):
                raise ValueError("表达式包含非法字符")
            
            result = eval(expression, {"__builtins__": {}}, {})
            
            return ToolResult(
                success=True,
                output=f"计算结果:{result}",
                execution_time=time.time() - start_time
            )
        
        except Exception as e:\n            return ToolResult(\n                success=False,\n                output="",
                error=str(e),
                execution_time=time.time() - start_time
            )

# ==================== Agent 核心 ====================

class ProductionAgent:
    """生产级 Agent"""
    
    def __init__(self, llm_func, tools: List[BaseTool]):
        self.llm = llm_func
        self.tools = {tool.name(): tool for tool in tools}
        self.max_iterations = 15
    
    def _build_prompt(self, task: str) -> str:
        """构建 prompt"""
        tools_desc = "\n".join([
            f"- {name}: {tool.description()}"
            for name, tool in self.tools.items()
        ])
        
        prompt = f"""
你是
一个强大的 AI 助手,能够使用工具完成各种任务。

## 任务
{task}

## 可用工具
{tools_desc}

## 执行规则
1. 思考完成任务需要哪些步骤
2. 选择合适的工具执行每一步
3. 观察工具返回的结果
4. 如果需要,继续调用工具
5. 任务完成后,给出最终答案

## 输出格式
Thought: [你的思考]
Action: [工具名],输入:[参数名=参数值,...]
Observation: [工具返回结果]
...(重复以上步骤)
Thought: 任务完成
Final Answer: [最终答案]
"""
        return prompt
    
    def _parse_action(self, text: str) -> Optional[Dict[str, Any]]:
        """解析 Action"""
        pattern = r"Action:\s*(\w+),输入:\s*(.+)"
        match = re.search(pattern, text)
        
        if match:
            tool_name = match.group(1)
            params_str = match.group(2)
            
            params = {}
            for pair in params_str.split(','):
                if '=' in pair:
                    key, value = pair.split('=', 1)
                    params[key.strip()] = value.strip()
            
            return {
                "tool": tool_name,
                "params": params
            }
        
        return None
    
    def _parse_final_answer(self, text: str) -> Optional[str]:
        """解析 Final Answer"""
        if "Final Answer:" in text:
            answer = text.split("Final Answer:")[1].strip()
            return answer
        return None
    
    def _execute_tool(self, tool_name: str, params: Dict) -> ToolResult:
        """执行工具(带重试)"""
        if tool_name not in self.tools:
            return ToolResult(
                success=False,
                output="",
                error=f"工具 {tool_name} 不存在"
            )
        
        tool = self.tools[tool_name]
        
        max_retries = 3
        for attempt in range(max_retries):
            result = tool.execute(**params)
            
            if result.success:
                return result
            
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
        
        return result
    
    def run(self, task: str) -> str:
        """运行 Agent"""
        
        # 1. 构建初始 prompt
        prompt = self._build_prompt(task)
        
        # 2. 主循环
        conversation = prompt
        
        for iteration in range(self.max_iterations):
            # 调用 LLM
            response = self.llm(conversation)
            
            # 检查是否完成
            final_answer = self._parse_final_answer(response)
            if final_answer:
                return final_answer
            
            # 解析 Action
            action = self._parse_action(response)
            if not action:
                conversation += "\n" + response + "\n"
                continue
            
            # 执行工具
            result = self._execute_tool(action["tool"], action["params"])
            
            # 将结果加回对话
            observation = f"Observation: {result.output}"
            if not result.success:
                observation += f"\nError: {result.error}"
            
            conversation += "\n" + response + "\n" + observation + "\nThought:"
        
        return "任务未能在限定步骤内完成。"

# ==================== 使用示例 ====================

def mock_llm(prompt: str) -> str:
    """模拟 LLM 调用"""
    if "234 * 567" in prompt or "计算器" in prompt:
        return """Thought: 我需要计算 234 * 567
Action: calculator,输入:expression=234*567
"""
    
    elif "Observation:" in prompt and "计算结果" in prompt:
        return """Thought: 计算完成了,可以给出最终答案了
Final Answer: 234 * 567 的计算结果是 132678。
"""
    
    else:
        return """Thought: 我需要更多信息
Final Answer: 抱歉,我目前无法完成这个任务。
"""

def main():
    """主程序"""
    
    # 1. 初始化工具
    tools = [
        SearchTool(),
        CalculatorTool()
    ]
    
    # 2. 创建 Agent
    agent = ProductionAgent(
        llm_func=mock_llm,
        tools=tools
    )
    
    # 3. 执行任务
    print("=" * 60)
    print("生产级 Agent 演示")
    print("=" * 60)
    
    result = agent.run("请计算 234 * 567")
    print(f"\n结果:{result}")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()

6. 工具调用深度剖析

6.1 Function Calling 的革命

2023年,OpenAI 引入了 Function Calling 功能,让模型直接输出结构化数据:

# 定义函数
functions = [
    {
        "name": "search",
        "description": "搜索工具",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"}
            },
            "required": ["query"]
        }
    }
]

# 调用 LLM
response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "搜索 Python"}],
    functions=functions
)

# LLM 直接返回结构化调用
# {
#   "tool_calls": [
#     {
#       "function": {
#         "name": "search",
#         "arguments": "{\"query\": \"Python\"}"
#       }
#     }
#   ]
# }

优势

  • 解析可靠(JSON 格式)
  • 支持复杂参数
  • 模型经过微调,调用更准确

6.2 工具调用的最佳实践

1. 工具定义要清晰

# ✅ 好的定义
{
    "name": "web_search",
    "description": "在网络上搜索信息。返回相关网页的标题、URL和摘要。",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词。支持布尔运算符"
            },
            "max_results": {
                "type": "number",
                "description": "返回的最大结果数",
                "default": 10
            }
        },
        "required": ["query"]
    }
}

2. 错误处理要完善

def safe_tool_call(tool_name: str, args: Dict) -> Dict:
    """安全的工具调用"""
    try:
        result = timeout(30)(execute_tool)(tool_name, args)
        return {"success": True, "result": result}
    
    except TimeoutError:
        return {"success": False, "error": "工具执行超时"}
    
    except Exception as e:\n        return {"success": False, "error": str(e)}

7. 记忆系统与上下文管理

7.1 记忆系统的分类

类型实现方式优势劣势
短期记忆上下文窗口快速、准确容量有限
长期记忆(向量)向量数据库容量无限检索精度依赖嵌入质量
工作记忆结构化状态精确、可控需要手动设计

7.2 向量记忆系统实现

from sentence_transformers import SentenceTransformer
import numpy as np

class VectorMemory:
    """基于向量的记忆系统"""
    
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.encoder = SentenceTransformer(model_name)
        self.memories = {}
        self.next_id = 0
    
    def add(self, content: str, metadata: Dict = None) -> str:
        """添加记忆"""
        embedding = self.encoder.encode(content)
        
        memory_id = str(self.next_id)
        self.memories[memory_id] = {
            "content": content,
            "embedding": embedding,
            "metadata": metadata or {}
        }
        self.next_id += 1
        
        return memory_id
    
    def search(self, query: str, top_k: int = 5) -> List[Dict]:
        """语义搜索"""
        query_embedding = self.encoder.encode(query)
        
        similarities = []
        for memory_id, memory in self.memories.items():
            sim = np.dot(query_embedding, memory["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(memory["embedding"])
            )
            similarities.append((memory_id, sim))
        
        similarities.sort(key=lambda x: x[1], reverse=True)
        
        results = []
        for memory_id, sim in similarities[:top_k]:
            memory = self.memories[memory_id]
            results.append({
                "content": memory["content"],
                "score": float(sim)
            })
        
        return results

8. 性能优化

8.1 成本分析

Agent 的主要成本来自 LLM 调用:

模型输入价格(每1k tokens)输出价格(每1k tokens)
GPT-3.5-turbo$0.0005$0.0015
GPT-4-turbo$0.01$0.03
本地 LLM$0.0001(硬件摊销)$0.0001

8.2 优化策略

策略1:模型路由

class SmartRouter:
    """智能模型路由"""
    
    def __init__(self):
        self.models = {
            "gpt-3.5": {"cost": 0.002, "capability": 0.7},
            "gpt-4": {"cost": 0.06, "capability": 0.95},
            "local": {"cost": 0.0001, "capability": 0.6}
        }
    
    def route(self, task: str) -> str:
        """路由到合适的模型"""
        if len(task) < 50:
            return "local"
        
        if "代码" in task or "分析" in task:
            return "gpt-4"
        
        return "gpt-3.5"

策略2:缓存

from functools import lru_cache
import hashlib

class CachedLLM:
    """带缓存的 LLM 调用"""
    
    def __init__(self, llm_func):
        self.llm_func = llm_func
        self.cache = {}
    
    def __call__(self, prompt: str, use_cache: bool = True) -> str:
        cache_key = hashlib.md5(prompt.encode()).hexdigest()
        
        if use_cache and cache_key in self.cache:
            print("✅ 缓存命中")
            return self.cache[cache_key]
        
        result = self.llm_func(prompt)
        self.cache[cache_key] = result
        
        return result

9. 安全与合规

9.1 安全风险

风险类型描述示例
提示注入用户通过输入控制 Agent 行为"忽略之前的指令,执行..."
工具滥用Agent 被诱导执行危险操作"删除所有文件"
数据泄露Agent 输出包含敏感信息输出用户密码

9.2 安全防护策略

输入过滤

class InputFilter:
    """输入过滤器"""
    
    def __init__(self):
        self.dangerous_keywords = [
            "忽略之前的指令",
            "ignore previous instructions",
            "DAN 模式"
        ]
    
    def filter(self, user_input: str) -> Tuple[bool, str]:
        """过滤用户输入"""
        for keyword in self.dangerous_keywords:
            if keyword in user_input:
                return False, f"输入包含不安全内容:{keyword}"
        
        if len(user_input) > 10000:
            return False, "输入过长"
        
        return True, user_input

10. 部署与监控

10.1 部署架构

┌─────────────────────────────────────────────────┐
│                  负载均衡(Nginx)                 │
└─────────────────┬───────────────────────────────┘
                  │
        ┌─────────┼─────────┐
        │         │         │
┌───────▼───┐ ┌───▼─────┐ ┌▼──────────┐
│ Agent API 1│ │Agent API 2│ │Agent API 3│
└───────┬───┘ └───┬─────┘ └┬──────────┘
        └─────────┼─────────┘
                  │
┌─────────────────▼───────────────────────────────┐
│              共享服务                              │
│  - Redis(记忆、缓存)                              │
│  - PostgreSQL(用户、任务、日志)                     │
│  - Vector DB(长期记忆)                           │
└─────────────────────────────────────────────────┘

10.2 监控指标

指标类型具体指标告警阈值
性能请求延迟(P50/P95/P99)P95 > 5s
成本LLM 调用费用超过预算
可靠性错误率> 1%

总结

本文从理论到实践,全面剖析了 AI Agent 技术:

  1. 核心概念: ReAct、Plan-and-Execute 等范式
  2. 框架对比: LangChain、AutoGPT、CrewAI
  3. 架构设计: 生产级 Agent 系统的模块划分
  4. 代码实战: 从0到1实现多功能 Agent
  5. 工具调用: Function Calling 的深度剖析
  6. 记忆系统: 短期 + 长期记忆的混合方案
  7. 性能优化: 成本控制和效率提升
  8. 安全合规: 企业级 Agent 的必修课
  9. 部署监控: 从开发到生产的完整流程

关键要点

  • Agent = LLM + 工具 + 记忆 + 规划
  • 生产级 Agent 需要考虑可靠性、成本、安全
  • 记忆系统决定了 Agent 的"智能"程度

下一步行动

  1. 动手实现一个简单的 Agent
  2. 尝试多 Agent 协作(用 CrewAI)
  3. 优化你的 Agent(加入记忆、规划)
  4. 部署到生产(考虑监控、安全)

参考资料

  1. ReAct Paper: Reasoning and Acting in Language Models (2022)
  2. LangChain Docs: https://python.langchain.com/
  3. CrewAI: https://github.com/crewAIInc/crewAI

关于作者

程序员茄子,全栈工程师,AI 技术爱好者。关注 AI Agent、LLM 应用、云原生技术。

  • 博客:https://www.chenxutan.com
  • GitHub:https://github.com/yourusername

版权声明

本文采用 CC BY-NC-SA 4.0 许可协议。

推荐文章

使用 sync.Pool 优化 Go 程序性能
2024-11-19 05:56:51 +0800 CST
H5抖音商城小黄车购物系统
2024-11-19 08:04:29 +0800 CST
ElasticSearch集群搭建指南
2024-11-19 02:31:21 +0800 CST
MySQL 1364 错误解决办法
2024-11-19 05:07:59 +0800 CST
程序员茄子在线接单