在 AI 应用开发领域，Agent 已经成为连接大语言模型与实际业务需求的关键技术。它不再只是简单调用 API 返回文本，而是能够理解复杂任务、制定执行计划、使用工具并完成多步骤操作的智能体。对于想要从基础 Python 语法过渡到实际 AI 应用开发的工程师来说，掌握 Agent 开发意味着能够构建真正具备自主决策能力的 AI 系统。

本文将从零开始构建一个完整的 AI Agent 项目，涵盖环境准备、核心概念理解、框架选型、代码实现到生产部署的全流程。不同于简单演示，我们会深入解释每个决策背后的技术考量，包括为什么选择特定框架、配置参数的含义、常见错误的根本原因以及如何设计可维护的 Agent 架构。

1. 理解 AI Agent 的核心工作机制

1.1 Agent 与传统程序的区别

传统程序执行的是预设逻辑流程，而 AI Agent 的核心能力在于动态决策。当用户提出“分析上周销售数据并生成报告”这样的复杂需求时，Agent 需要自主分解为多个子任务：获取数据、清洗整理、分析趋势、选择图表类型、生成文字描述。这种任务分解和工具调用的能力，让 Agent 能够处理开放式问题而非固定流程。

关键区别在于，Agent 内部维护着决策状态。它会根据当前执行结果决定下一步动作，比如数据获取失败时会尝试备用方案，而不是直接报错退出。这种状态管理能力使得 Agent 能够处理真实世界中的不确定性和异常情况。

1.2 LangChain 与 LangGraph 的架构差异

LangChain 提供了构建 Agent 所需的基础组件：模型封装、提示模板、记忆存储和工具调用。它的核心是 AgentExecutor，采用 while 循环不断决策直到任务完成。这种架构适合相对线性的任务流程，但在处理复杂工作流时显得不够灵活。

# LangChain 传统 Agent 执行流程
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool

# 定义工具
tools = [
    Tool(
        name="Search",
        func=search_api,
        description="用于搜索最新信息"
    )
]

# 创建 Agent
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 执行任务
result = agent_executor.invoke({"input": "查询北京今天的天气"})

LangGraph 引入了图结构来描述 Agent 的工作流，每个节点代表一个处理步骤，边代表状态转移条件。这种架构特别适合需要条件分支、循环执行或并行处理的任务。例如，一个客服 Agent 可能需要先判断用户意图，再根据意图类型选择不同的处理分支。

# LangGraph 的状态图定义
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], "消息历史"]
    current_step: Annotated[str, "当前步骤"]
    
def route_message(state: AgentState):
    last_message = state["messages"][-1]
    if "天气" in last_message.content:
        return "weather_agent"
    elif "新闻" in last_message.content:
        return "news_agent"
    else:
        return "general_agent"

# 构建图结构
builder = StateGraph(AgentState)
builder.add_node("router", route_message)
builder.add_node("weather_agent", process_weather)
builder.add_node("news_agent", process_news)
builder.set_entry_point("router")
builder.add_conditional_edges("router", route_message)
graph = builder.compile()

1.3 Transformer 模型在 Agent 中的角色

Transformer 架构是现代 AI Agent 的核心技术基础。它的自注意力机制能够理解长文本中的复杂关系，这对于 Agent 分析多步骤任务至关重要。在 Agent 系统中，Transformer 主要承担三个角色：

• 任务理解 ：解析用户输入的复杂需求，识别关键信息和隐含约束
• 决策生成 ：根据当前状态和可用工具，生成下一步行动计划
• 结果合成 ：将工具执行结果整合成自然语言响应

理解 Transformer 的工作原理有助于优化 Agent 的提示工程和错误处理。比如，当 Agent 频繁出现逻辑错误时，可能是由于注意力机制未能正确捕捉关键约束条件，需要通过改进提示词或增加示例来引导模型关注重点信息。

2. 开发环境准备与依赖管理

2.1 Python 环境配置最佳实践

Agent 开发对 Python 环境稳定性要求较高，推荐使用 conda 或 pyenv 进行环境隔离。不同版本的 LangChain 和模型库可能存在兼容性问题，固定环境能够确保部署一致性。

# 使用 conda 创建专用环境
conda create -n ai-agent python=3.11
conda activate ai-agent

# 安装核心依赖
pip install langchain langchain-community langgraph
pip install openai anthropic  # 根据使用的模型选择

# 开发工具
pip install jupyter ipython black flake8 mypy

对于 IDE 选择，VS Code 配合 Python 扩展和 Jupyter 支持能够提供良好的开发体验。关键配置包括设置正确的 Python 解释器路径、启用代码自动完成和类型检查。

// VS Code settings.json 相关配置
{
    "python.defaultInterpreterPath": "~/miniconda3/envs/ai-agent/bin/python",
    "python.analysis.autoImportCompletions": true,
    "python.analysis.typeCheckingMode": "basic"
}

2.2 模型 API 密钥安全管理

Agent 项目通常需要访问多个外部 API，如 OpenAI、Anthropic 或本地部署的模型服务。硬编码密钥是严重的安全隐患，必须通过环境变量或配置文件管理。

# 安全配置管理示例
import os
from dataclasses import dataclass

@dataclass
class ModelConfig:
    openai_api_key: str = os.getenv("OPENAI_API_KEY")
    anthropic_api_key: str = os.getenv("ANTHROPIC_API_KEY")
    base_url: str = os.getenv("MODEL_BASE_URL", "https://api.openai.com/v1")
    
    def validate(self):
        if not self.openai_api_key and not self.anthropic_api_key:
            raise ValueError("至少需要配置一个模型 API 密钥")

# 使用配置
config = ModelConfig()
config.validate()

对于生产环境，建议使用专门的密钥管理服务（如 AWS Secrets Manager、HashiCorp Vault）或平台提供的安全配置功能。本地开发时可以在 .env 文件中设置环境变量，但务必确保该文件被添加到 .gitignore 中。

2.3 依赖版本锁定策略

AI 库更新频繁，但盲目升级可能导致兼容性问题。建议使用 requirements.txt 精确锁定版本，并在独立环境中测试升级。

# requirements.txt
langchain==0.1.0
langchain-community==0.0.10
langgraph==0.0.30
openai==1.3.0
anthropic==0.7.2
python-dotenv==1.0.0

定期更新依赖时，应该先在测试环境验证兼容性，特别关注 breaking changes。LangChain 的版本更新通常会在文档中说明重大变更，升级前需要仔细阅读。

3. 构建第一个可运行的 AI Agent

3.1 设计简单的工具调用 Agent

我们从最简单的 ReAct Agent 开始，让模型能够使用计算器和搜索引擎工具。这种设计模式是大多数复杂 Agent 的基础框架。

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain import hub
from langchain_openai import ChatOpenAI
import math

# 定义工具函数
def calculator(expression: str) -> str:
    """计算数学表达式，支持 + - * / 和基本函数"""
    try:
        # 安全评估数学表达式
        allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("_")}
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

def search_web(query: str) -> str:
    """模拟网络搜索功能"""
    # 实际项目中这里会接入真正的搜索 API
    return f"搜索 '{query}' 的结果: 示例信息"

# 创建工具列表
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="用于数学计算，输入数学表达式如 '2 + 3 * 4'"
    ),
    Tool(
        name="WebSearch",
        func=search_web,
        description="用于搜索最新信息，输入搜索关键词"
    )
]

# 从 LangChain Hub 获取优化过的提示模板
prompt = hub.pull("hwchase17/react")

# 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 创建 Agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

3.2 实现完整的执行流程

有了 Agent 定义后，我们需要实现完整的交互流程，包括输入处理、执行监控和结果解析。

def run_agent_interactive():
    """交互式运行 Agent"""
    print("AI Agent 已启动，输入 '退出' 结束对话")
    
    while True:
        try:
            user_input = input("\n用户: ").strip()
            if user_input.lower() in ['退出', 'exit', 'quit']:
                print("对话结束")
                break
                
            if not user_input:
                continue
                
            # 执行 Agent
            result = agent_executor.invoke({"input": user_input})
            print(f"\nAgent: {result['output']}")
            
        except KeyboardInterrupt:
            print("\n程序被用户中断")
            break
        except Exception as e:
            print(f"执行错误: {str(e)}")
            # 可以根据错误类型提供更友好的提示

if __name__ == "__main__":
    # 验证环境配置
    if not os.getenv("OPENAI_API_KEY"):
        print("错误: 请设置 OPENAI_API_KEY 环境变量")
        exit(1)
        
    run_agent_interactive()

3.3 验证 Agent 的核心能力

运行上述代码后，我们可以通过几个典型测试用例验证 Agent 是否正常工作：

# 测试用例
test_cases = [
    "计算 15 的平方根是多少？",
    "搜索人工智能的最新发展",
    "先计算 100 除以 3 的结果，然后搜索这个结果的近似值"
]

for i, case in enumerate(test_cases, 1):
    print(f"\n测试用例 {i}: {case}")
    result = agent_executor.invoke({"input": case})
    print(f"结果: {result['output']}")

正常运行的 Agent 应该能够：

• 正确识别需要使用的工具
• 生成适当的参数调用工具
• 将工具结果整合成自然语言响应
• 处理多步骤任务时保持上下文连贯

4. Agent 核心组件深度配置

4.1 模型参数调优策略

选择适合的模型参数对 Agent 性能影响巨大。temperature 控制创造性，top_p 控制词汇选择范围，max_tokens 限制响应长度。

# 针对不同任务类型的模型配置
class ModelConfigPresets:
    """模型配置预设"""
    
    @staticmethod
    def get_analytical_preset():
        """分析型任务配置：低随机性，注重准确性"""
        return {
            "temperature": 0.1,
            "top_p": 0.9,
            "max_tokens": 2000,
            "frequency_penalty": 0.1  # 降低重复内容
        }
    
    @staticmethod
    def get_creative_preset():
        """创意型任务配置：适度随机性，鼓励多样性"""
        return {
            "temperature": 0.7,
            "top_p": 0.95,
            "max_tokens": 1000,
            "presence_penalty": 0.2  # 鼓励新话题
        }
    
    @staticmethod
    def get_conversational_preset():
        """对话型任务配置：平衡准确性和自然度"""
        return {
            "temperature": 0.3,
            "top_p": 0.92,
            "max_tokens": 800,
            "frequency_penalty": 0.05
        }

# 应用配置示例
analytical_llm = ChatOpenAI(
    model="gpt-4",
    **ModelConfigPresets.get_analytical_preset()
)

4.2 工具设计的核心原则

工具是 Agent 能力的扩展，良好的工具设计应该遵循单一职责原则和明确的接口契约。

from typing import Type, Any
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    """计算器工具的输入模型"""
    expression: str = Field(description="数学表达式，如 '2 + 3 * 4'")

class AdvancedCalculatorTool:
    """增强版计算器工具，支持输入验证"""
    
    def __init__(self):
        self.name = "AdvancedCalculator"
        self.description = "执行数学计算，支持基本运算和常用函数"
        self.args_schema: Type[BaseModel] = CalculatorInput
    
    def _validate_expression(self, expression: str) -> bool:
        """验证数学表达式安全性"""
        # 允许的字符和函数
        allowed_chars = set("0123456789+-*/.() ")
        allowed_functions = {'sqrt', 'sin', 'cos', 'tan', 'log', 'exp'}
        
        # 基础字符检查
        if not all(c in allowed_chars or c.isalpha() for c in expression):
            return False
            
        # 函数调用安全检查
        words = expression.split()
        for word in words:
            if word.isalpha() and word not in allowed_functions:
                return False
                
        return True
    
    def __call__(self, expression: str) -> str:
        if not self._validate_expression(expression):
            return "错误: 表达式包含不安全字符或函数"
            
        try:
            allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("_")}
            result = eval(expression, {"__builtins__": {}}, allowed_names)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"

# 使用增强工具
advanced_calc = AdvancedCalculatorTool()
tools.append(Tool(
    name=advanced_calc.name,
    func=advanced_calc,
    description=advanced_calc.description,
    args_schema=advanced_calc.args_schema
))

4.3 记忆管理机制实现

Agent 的记忆能力决定了其处理长对话和复杂任务的能力。LangChain 提供了多种记忆后端，从简单的内存存储到持久化数据库。

from langchain.memory import ConversationBufferWindowMemory, ChatMessageHistory
from langchain.schema import BaseChatMessageHistory

class PersistentMemoryManager:
    """持久化记忆管理器"""
    
    def __init__(self, storage_path: str = "memory_storage"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
    
    def get_memory_for_session(self, session_id: str) -> ConversationBufferWindowMemory:
        """获取或创建会话记忆"""
        file_path = self.storage_path / f"{session_id}.json"
        
        if file_path.exists():
            # 从文件加载历史记录
            with open(file_path, 'r', encoding='utf-8') as f:
                history_data = json.load(f)
        else:
            history_data = []
            
        # 创建记忆实例
        message_history = ChatMessageHistory()
        for msg in history_data:
            if msg['type'] == 'human':
                message_history.add_user_message(msg['content'])
            else:
                message_history.add_ai_message(msg['content'])
                
        return ConversationBufferWindowMemory(
            chat_memory=message_history,
            return_messages=True,
            k=10  # 保留最近10轮对话
        )
    
    def save_memory(self, session_id: str, memory: ConversationBufferWindowMemory):
        """保存会话记忆"""
        file_path = self.storage_path / f"{session_id}.json"
        messages = memory.chat_memory.messages
        
        history_data = []
        for msg in messages:
            history_data.append({
                'type': 'human' if msg.type == 'human' else 'ai',
                'content': msg.content,
                'timestamp': datetime.now().isoformat()
            })
            
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(history_data, f, ensure_ascii=False, indent=2)

# 使用记忆增强的 Agent
memory_manager = PersistentMemoryManager()
session_memory = memory_manager.get_memory_for_session("user_123")

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    memory=session_memory,
    handle_parsing_errors=True
)

5. 生产环境部署与监控

5.1 错误处理与重试机制

生产环境中的 Agent 必须能够优雅处理各种异常情况，包括网络超时、API 限制、无效输入等。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import openai

class RobustAgentExecutor:
    """增强鲁棒性的 Agent 执行器"""
    
    def __init__(self, base_executor: AgentExecutor, max_retries: int = 3):
        self.executor = base_executor
        self.max_retries = max_retries
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        retry=retry_if_exception_type((
            openai.APITimeoutError,
            openai.APIConnectionError
        ))
    )
    def invoke_with_retry(self, input_data: dict) -> dict:
        """带重试机制的调用方法"""
        try:
            return self.executor.invoke(input_data)
        except openai.RateLimitError:
            # 速率限制错误需要更长的等待
            time.sleep(60)
            raise
        except openai.InvalidRequestError as e:
            # 无效请求错误不应重试
            return {"output": f"请求参数错误: {str(e)}", "error": True}
        except Exception as e:
            # 其他错误记录日志并返回友好提示
            logging.error(f"Agent 执行错误: {str(e)}")
            return {"output": "系统暂时不可用，请稍后重试", "error": True}
    
    def safe_invoke(self, input_data: dict, fallback_response: str = None) -> dict:
        """安全调用方法，确保始终返回有效响应"""
        for attempt in range(self.max_retries):
            try:
                result = self.invoke_with_retry(input_data)
                if not result.get("error"):
                    return result
            except Exception as e:
                logging.warning(f"第 {attempt + 1} 次尝试失败: {str(e)}")
                if attempt == self.max_retries - 1:
                    break
                time.sleep(2 ** attempt)  # 指数退避
        
        # 所有重试都失败后返回降级响应
        return {
            "output": fallback_response or "服务暂时不可用，请稍后重试",
            "error": True,
            "fallback": True
        }

# 使用增强执行器
robust_executor = RobustAgentExecutor(agent_executor)
result = robust_executor.safe_invoke({"input": "重要查询"}, "系统维护中，请稍后访问")

5.2 性能监控与日志记录

完善的监控体系是生产环境 Agent 的必备条件，需要记录关键指标和异常信息。

import time
import logging
from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class PerformanceMetrics:
    """性能指标数据类"""
    start_time: float
    end_time: float = 0
    token_usage: Dict[str, int] = None
    tool_calls: int = 0
    success: bool = True
    
    @property
    def duration(self) -> float:
        return self.end_time - self.start_time
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "duration": self.duration,
            "token_usage": self.token_usage or {},
            "tool_calls": self.tool_calls,
            "success": self.success
        }

class MonitoringAgentExecutor:
    """带监控功能的 Agent 执行器"""
    
    def __init__(self, base_executor: AgentExecutor):
        self.executor = base_executor
        self.logger = logging.getLogger("agent.monitoring")
    
    def invoke(self, input_data: dict) -> dict:
        metrics = PerformanceMetrics(start_time=time.time())
        
        try:
            result = self.executor.invoke(input_data)
            metrics.end_time = time.time()
            metrics.success = True
            
            # 记录成功日志
            self.logger.info(
                "Agent 执行成功",
                extra={
                    "metrics": metrics.to_dict(),
                    "input": input_data.get("input", "")[:100],  # 截断长文本
                    "output_length": len(result.get("output", ""))
                }
            )
            
            return result
            
        except Exception as e:
            metrics.end_time = time.time()
            metrics.success = False
            
            # 记录错误日志
            self.logger.error(
                f"Agent 执行失败: {str(e)}",
                extra={
                    "metrics": metrics.to_dict(),
                    "input": input_data.get("input", "")[:100],
                    "error_type": type(e).__name__
                },
                exc_info=True
            )
            raise

# 配置结构化日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('agent.log'),
        logging.StreamHandler()
    ]
)

5.3 配置管理与环境适配

不同环境（开发、测试、生产）需要不同的配置，通过配置管理实现环境无缝切换。

from enum import Enum
import yaml

class Environment(Enum):
    DEVELOPMENT = "dev"
    TESTING = "test"
    PRODUCTION = "prod"

class ConfigManager:
    """配置管理器"""
    
    def __init__(self, env: Environment = Environment.DEVELOPMENT):
        self.env = env
        self.config = self._load_config()
    
    def _load_config(self) -> dict:
        """加载环境配置"""
        base_config = {
            "model": {
                "default": "gpt-3.5-turbo",
                "fallback": "gpt-3.5-turbo"
            },
            "timeouts": {
                "api_timeout": 30,
                "max_execution_time": 120
            },
            "logging": {
                "level": "INFO",
                "file": "agent.log"
            }
        }
        
        # 环境特定配置
        env_configs = {
            Environment.DEVELOPMENT: {
                "model": {"default": "gpt-3.5-turbo"},
                "timeouts": {"api_timeout": 60},
                "logging": {"level": "DEBUG"}
            },
            Environment.TESTING: {
                "model": {"default": "gpt-3.5-turbo"},
                "timeouts": {"api_timeout": 30},
                "logging": {"level": "INFO"}
            },
            Environment.PRODUCTION: {
                "model": {"default": "gpt-4", "fallback": "gpt-3.5-turbo"},
                "timeouts": {"api_timeout": 15, "max_execution_time": 60},
                "logging": {"level": "WARNING", "file": "/var/log/agent.log"}
            }
        }
        
        # 合并配置
        env_config = env_configs.get(self.env, {})
        return self._deep_merge(base_config, env_config)
    
    def _deep_merge(self, base: dict, update: dict) -> dict:
        """深度合并字典"""
        result = base.copy()
        for key, value in update.items():
            if isinstance(value, dict) and key in result and isinstance(result[key], dict):
                result[key] = self._deep_merge(result[key], value)
            else:
                result[key] = value
        return result
    
    def get(self, key: str, default=None):
        """获取配置值"""
        keys = key.split('.')
        value = self.config
        for k in keys:
            value = value.get(k, {})
        return value if value != {} else default

# 使用配置管理
config = ConfigManager(Environment.PRODUCTION)
llm = ChatOpenAI(
    model=config.get("model.default"),
    timeout=config.get("timeouts.api_timeout")
)

6. 常见问题排查与优化

6.1 Agent 典型错误分析

在实际开发中，Agent 可能遇到各种问题，需要系统化的排查方法。

问题现象	可能原因	检查步骤	解决方案
Agent 陷入循环	终止条件不明确	检查提示词中的停止条件	增加明确的完成标准
工具选择错误	工具描述不清晰	验证工具描述准确性	重写工具描述，增加示例
解析错误	模型输出格式不符	检查模型响应格式	改进输出解析逻辑
执行超时	任务过于复杂	分析任务分解合理性	增加步骤限制或简化任务

6.2 性能优化策略

随着任务复杂度增加，Agent 性能可能成为瓶颈，需要针对性优化。

class AgentOptimizer:
    """Agent 性能优化器"""
    
    @staticmethod
    def optimize_prompt_length(prompt: str, max_tokens: int = 4000) -> str:
        """优化提示词长度，避免超出模型限制"""
        # 估算 token 数量（简单版本）
        estimated_tokens = len(prompt) // 4
        
        if estimated_tokens > max_tokens:
            # 截断过长的提示词，保留关键部分
            lines = prompt.split('\n')
            important_sections = [line for line in lines if any(
                keyword in line for keyword in ['重要', '必须', '步骤', '规则']
            )]
            optimized = '\n'.join(important_sections[-10:])  # 保留最后10个重要部分
            return optimized + "\n...（内容已截断）"
        
        return prompt
    
    @staticmethod
    def enable_streaming(llm, callback_func=None):
        """启用流式响应，改善用户体验"""
        return llm.bind(streaming=True, callback_func=callback_func)
    
    @staticmethod
    def setup_caching():
        """设置缓存机制，减少重复计算"""
        from langchain.cache import InMemoryCache
        from langchain.globals import set_llm_cache
        
        set_llm_cache(InMemoryCache())

# 应用优化
AgentOptimizer.setup_caching()
optimized_llm = AgentOptimizer.enable_streaming(llm)

6.3 安全加固措施

AI Agent 涉及用户输入处理和外部工具调用，必须考虑安全问题。

import re

class SecurityValidator:
    """安全验证器"""
    
    @staticmethod
    def validate_input(text: str, max_length: int = 1000) -> bool:
        """验证用户输入安全性"""
        if len(text) > max_length:
            return False
            
        # 检查潜在恶意模式
        malicious_patterns = [
            r"\.\./",  # 路径遍历
            r";\s*(rm|drop|delete)",  # 命令注入
            r"<script[^>]*>",  # XSS 攻击
        ]
        
        for pattern in malicious_patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return False
                
        return True
    
    @staticmethod
    def sanitize_tool_output(output: str) -> str:
        """清理工具输出，防止信息泄露"""
        # 移除敏感信息模式
        sensitive_patterns = [
            r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",  # 信用卡号
            r"\b\d{3}[- ]?\d{2}[- ]?\d{4}\b",  # 社会安全号
        ]
        
        sanitized = output
        for pattern in sensitive_patterns:
            sanitized = re.sub(pattern, "[敏感信息已屏蔽]", sanitized)
            
        return sanitized

# 在 Agent 调用前加入安全验证
def safe_agent_invoke(agent, input_data: dict) -> dict:
    user_input = input_data.get("input", "")
    
    if not SecurityValidator.validate_input(user_input):
        return {"output": "输入内容包含不安全字符，请重新输入", "error": True}
    
    result = agent.invoke(input_data)
    
    if "output" in result:
        result["output"] = SecurityValidator.sanitize_tool_output(result["output"])
    
    return result

构建生产级 AI Agent 需要综合考虑架构设计、错误处理、性能监控和安全防护。从简单的工具调用开始，逐步增加记忆能力、优化提示工程、完善监控体系，最终形成能够可靠处理真实需求的智能体系统。实际项目中还需要根据具体业务场景调整工具集和决策逻辑，但本文提供的框架和最佳实践能够为大多数 Agent 开发项目奠定坚实基础。