在AI Agent开发领域Claude Fable 5的出现标志着自主智能体技术迈入了新的里程碑。作为Anthropic推出的Mythos级模型它专为复杂的知识工作和编程任务设计具备强大的推理能力和1M token的上下文窗口。本文将深入探讨如何基于Claude Fable 5构建名为Damon的终极AI Agent框架从核心原理到生产级实战为开发者提供完整的解决方案。1. Claude Fable 5与AI Agent技术背景1.1 Claude Fable 5核心特性解析Claude Fable 5是Anthropic在2026年6月发布的重要模型属于Mythos类别专门针对需要长时间运行、复杂且异步的任务场景。该模型支持文本、图像和文件输入输出为文本格式具备强大的推理支持能力。在实际应用中Claude Fable 5特别擅长处理那些原本需要人类花费数小时、数天甚至数周才能完成的端到端工作。从技术参数来看Claude Fable 5的输入token价格为每百万10美元输出token价格为每百万50美元这在同类模型中具有相当的竞争力。模型内置了自动验证循环和强大的安全防护机制能够在执行明确定义的任务时自动纠错大大减少了人工干预的需求。1.2 AI Agent框架的发展现状当前AI Agent开发领域正处于快速演进阶段从早期的简单对话式AI发展到现在的自主智能体系统。传统的对话式AI主要关注单轮或有限轮次的交互而现代AI Agent则需要具备长期记忆、工具调用、任务分解和自主决策等复杂能力。Damon框架的设计目标就是填补现有AI Agent框架在复杂任务处理、长期运行稳定性以及工具生态整合方面的空白。通过结合Claude Fable 5的强大能力Damon框架能够支持从简单的自动化脚本到复杂的企业级业务流程自动化等多种应用场景。1.3 Damon框架的设计哲学Damon框架的核心设计理念是模块化、可扩展、生产就绪。与传统的AI Agent框架不同Damon强调将智能体的各个组件进行清晰的责任分离包括任务规划器、工具执行器、状态管理器和安全监控器等核心模块。这种设计使得开发者能够根据具体需求灵活地替换或扩展特定组件而不需要重写整个系统。框架还特别注重生产环境下的稳定性和可维护性内置了完善的日志记录、错误处理和性能监控机制。这些特性使得Damon框架特别适合需要7x24小时稳定运行的商业应用场景。2. 环境准备与开发工具链配置2.1 基础开发环境要求构建Damon框架需要准备以下基础环境操作系统Windows 10/11, macOS 12, 或 Ubuntu 20.04Python版本3.9或更高版本内存至少8GB推荐16GB以上存储空间至少10GB可用空间对于Python环境管理强烈建议使用conda或pyenv来创建独立的虚拟环境避免依赖冲突。# 使用conda创建虚拟环境 conda create -n damon-framework python3.9 conda activate damon-framework # 或者使用venv python -m venv damon-env source damon-env/bin/activate # Linux/macOS damon-env\Scripts\activate # Windows2.2 核心依赖库安装Damon框架依赖多个重要的Python库以下是核心依赖的安装命令pip install openai anthropic-sdk pydantic asyncio aiohttp pip install sqlalchemy alembic # 数据库支持 pip install pytest pytest-asyncio # 测试框架 pip install black isort flake8 # 代码格式化工具特别需要注意的是由于Claude Fable 5通过OpenRouter提供服务我们需要安装兼容OpenAI API的客户端库。OpenRouter的API与OpenAI API兼容这意味着大多数现有的OpenAI SDK都可以通过简单的配置修改来接入Claude Fable 5。2.3 OpenRouter API配置要使用Claude Fable 5首先需要在OpenRouter平台注册账号并获取API密钥。OpenRouter作为模型聚合平台提供了统一的API接口来访问多个提供商的模型服务。# config.py - OpenRouter配置 import os from typing import Optional class OpenRouterConfig: BASE_URL https://openrouter.ai/api/v1 MODEL_NAME anthropic/claude-fable-5 classmethod def get_api_key(cls) - str: api_key os.getenv(OPENROUTER_API_KEY) if not api_key: raise ValueError(OPENROUTER_API_KEY环境变量未设置) return api_key classmethod def get_headers(cls) - dict: return { Authorization: fBearer {cls.get_api_key()}, Content-Type: application/json, HTTP-Referer: https://your-domain.com, # 你的网站域名 X-Title: Damon AI Agent Framework # 应用名称 }2.4 开发工具和IDE配置推荐使用VS Code或PyCharm作为主要开发环境并安装以下扩展来提高开发效率Python扩展提供代码补全、调试支持Pylance增强的Python语言支持Black Formatter自动代码格式化GitLensGit版本控制增强在VS Code的settings.json中添加以下配置{ python.defaultInterpreterPath: ./damon-env/bin/python, editor.formatOnSave: true, python.formatting.provider: black, python.linting.enabled: true, python.linting.flake8Enabled: true }3. Damon框架核心架构设计3.1 模块化架构概述Damon框架采用分层架构设计将系统分为四个核心层次接口层、核心层、工具层和持久层。这种设计确保了各组件之间的松耦合和高内聚便于维护和扩展。接口层负责处理外部请求和响应格式转换支持REST API、WebSocket和命令行等多种接入方式。核心层包含任务调度、状态管理和安全控制等关键业务逻辑。工具层提供各种可插拔的功能模块如网络请求、文件操作、数据分析等。持久层负责数据的长期存储和检索。3.2 智能体状态管理设计在AI Agent系统中状态管理是确保长期任务正确执行的关键。Damon框架实现了基于事件溯源的状态管理机制能够完整记录智能体的决策过程和执行历史。# state_manager.py - 智能体状态管理 from typing import Dict, Any, List from datetime import datetime import json class AgentState: def __init__(self, agent_id: str): self.agent_id agent_id self.current_task: Optional[str] None self.task_history: List[Dict] [] self.context: Dict[str, Any] {} self.last_updated: datetime datetime.now() def add_task_event(self, event_type: str, data: Dict[str, Any]): event { timestamp: datetime.now().isoformat(), type: event_type, data: data, task: self.current_task } self.task_history.append(event) self.last_updated datetime.now() def to_dict(self) - Dict[str, Any]: return { agent_id: self.agent_id, current_task: self.current_task, task_history: self.task_history, context: self.context, last_updated: self.last_updated.isoformat() } class StateManager: def __init__(self): self.agents: Dict[str, AgentState] {} def get_agent_state(self, agent_id: str) - AgentState: if agent_id not in self.agents: self.agents[agent_id] AgentState(agent_id) return self.agents[agent_id] def save_state(self, agent_id: str, filepath: str): state self.get_agent_state(agent_id).to_dict() with open(filepath, w) as f: json.dump(state, f, indent2)3.3 工具调用系统实现工具调用是AI Agent能力的核心扩展机制。Damon框架实现了基于装饰器的工具注册系统使得开发者可以轻松地为智能体添加新的能力。# tool_system.py - 工具调用系统 import inspect from typing import Callable, Dict, Any, List from functools import wraps class ToolRegistry: def __init__(self): self.tools: Dict[str, Dict] {} def register_tool(self, name: str None, description: str ): def decorator(func: Callable) - Callable: tool_name name or func.__name__ # 提取函数参数信息 sig inspect.signature(func) parameters [] for param_name, param in sig.parameters.items(): param_info { name: param_name, type: param.annotation if param.annotation ! inspect.Parameter.empty else str, required: param.default inspect.Parameter.empty } parameters.append(param_info) self.tools[tool_name] { function: func, description: description, parameters: parameters } wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper return decorator def get_tool_schema(self) - List[Dict]: schemas [] for name, tool_info in self.tools.items(): schema { name: name, description: tool_info[description], parameters: { type: object, properties: {}, required: [] } } for param in tool_info[parameters]: schema[parameters][properties][param[name]] { type: self._python_type_to_json_type(param[type]) } if param[required]: schema[parameters][required].append(param[name]) schemas.append(schema) return schemas def _python_type_to_json_type(self, py_type) - str: type_map { str: string, int: integer, float: number, bool: boolean, list: array, dict: object } return type_map.get(py_type, string) # 工具使用示例 tool_registry ToolRegistry() tool_registry.register_tool( nameweb_search, description在互联网上搜索相关信息 ) def web_search(query: str, max_results: int 5) - List[Dict]: # 实现具体的搜索逻辑 return [{title: 示例结果, url: https://example.com, snippet: 相关摘要}] tool_registry.register_tool( namecalculate, description执行数学计算 ) def calculate(expression: str) - float: # 实现安全计算逻辑 return eval(expression) # 注意生产环境需要更安全的实现4. Claude Fable 5集成实战4.1 API客户端封装为了充分发挥Claude Fable 5的能力我们需要实现一个高效的API客户端。由于OpenRouter提供与OpenAI兼容的API我们可以基于现有的OpenAI SDK进行封装。# claude_client.py - Claude Fable 5客户端 import aiohttp import json from typing import List, Dict, Any, Optional from config import OpenRouterConfig class ClaudeFable5Client: def __init__(self): self.base_url OpenRouterConfig.BASE_URL self.headers OpenRouterConfig.get_headers() self.model OpenRouterConfig.MODEL_NAME async def create_chat_completion( self, messages: List[Dict[str, str]], temperature: float 0.7, max_tokens: int 4000, tools: Optional[List[Dict]] None ) - Dict[str, Any]: payload { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, } if tools: payload[tools] tools async with aiohttp.ClientSession() as session: async with session.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) as response: if response.status 200: return await response.json() else: error_text await response.text() raise Exception(fAPI请求失败: {response.status} - {error_text}) async def stream_chat_completion( self, messages: List[Dict[str, str]], temperature: float 0.7, max_tokens: int 4000 ): payload { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, stream: True } async with aiohttp.ClientSession() as session: async with session.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) as response: async for line in response.content: if line.startswith(bdata: ): data line[6:].strip() if data b[DONE]: break try: chunk json.loads(data) yield chunk except json.JSONDecodeError: continue4.2 智能体对话管理基于Claude Fable 5的对话管理需要处理多轮对话、上下文维护和工具调用等复杂场景。以下是核心的对话管理器实现# dialogue_manager.py - 对话管理器 from typing import List, Dict, Any, Optional from claude_client import ClaudeFable5Client from tool_system import ToolRegistry class DialogueManager: def __init__(self, tool_registry: ToolRegistry): self.client ClaudeFable5Client() self.tool_registry tool_registry self.conversation_history: List[Dict] [] async def process_message( self, user_message: str, agent_context: Dict[str, Any] None ) - Dict[str, Any]: # 构建消息历史 messages self._build_messages(user_message, agent_context) # 获取可用工具模式 tools self.tool_registry.get_tool_schema() # 调用Claude Fable 5 response await self.client.create_chat_completion( messagesmessages, toolstools if tools else None ) # 处理响应 return await self._handle_response(response, agent_context) def _build_messages(self, user_message: str, context: Dict None) - List[Dict]: messages [] # 添加系统提示 system_prompt self._build_system_prompt(context) messages.append({role: system, content: system_prompt}) # 添加对话历史 messages.extend(self.conversation_history[-10:]) # 最近10轮对话 # 添加当前用户消息 messages.append({role: user, content: user_message}) return messages def _build_system_prompt(self, context: Dict None) - str: base_prompt 你是一个专业的AI助手基于Claude Fable 5构建。你具备以下能力 1. 分析和解决复杂问题 2. 使用可用工具完成任务 3. 进行多步推理和规划 4. 在不确定的情况下主动澄清需求 请以专业、准确的方式回应用户请求。 if context and context.get(current_task): base_prompt f\n\n当前正在执行的任务: {context[current_task]} return base_prompt async def _handle_response( self, response: Dict[str, Any], context: Dict[str, Any] ) - Dict[str, Any]: choice response[choices][0] message choice[message] result { content: message.get(content, ), role: message[role], tool_calls: message.get(tool_calls, []) } # 处理工具调用 if result[tool_calls]: tool_results await self._execute_tool_calls(result[tool_calls]) result[tool_results] tool_results # 如果有工具调用结果需要再次调用模型进行总结 if tool_results: follow_up_response await self._process_tool_results( result, tool_results ) result[final_response] follow_up_response # 更新对话历史 self.conversation_history.append({role: user, content: 用户消息}) self.conversation_history.append(message) return result async def _execute_tool_calls(self, tool_calls: List[Dict]) - List[Dict]: results [] for tool_call in tool_calls: tool_name tool_call[function][name] arguments json.loads(tool_call[function][arguments]) tool_info self.tool_registry.tools.get(tool_name) if tool_info: try: result await tool_info[function](**arguments) results.append({ tool_name: tool_name, result: result, success: True }) except Exception as e: results.append({ tool_name: tool_name, error: str(e), success: False }) else: results.append({ tool_name: tool_name, error: f工具未找到: {tool_name}, success: False }) return results4.3 长上下文任务处理Claude Fable 5支持1M token的上下文窗口这使得处理长文档和复杂任务成为可能。以下是长上下文任务处理的优化实现# long_context_handler.py - 长上下文处理 from typing import List, Dict, Any import hashlib class LongContextHandler: def __init__(self, max_tokens: int 800000): # 保留200k token缓冲 self.max_tokens max_tokens def chunk_document(self, document: str, chunk_size: int 10000) - List[Dict]: 将长文档分块处理 chunks [] words document.split() current_chunk [] current_size 0 for word in words: if current_size len(word) 1 chunk_size: # 1 for space chunks.append({ content: .join(current_chunk), hash: hashlib.md5( .join(current_chunk).encode()).hexdigest() }) current_chunk [word] current_size len(word) else: current_chunk.append(word) current_size len(word) 1 if current_chunk: chunks.append({ content: .join(current_chunk), hash: hashlib.md5( .join(current_chunk).encode()).hexdigest() }) return chunks def build_context_summary(self, chunks: List[Dict]) - str: 构建上下文摘要用于在对话中引用 summary 文档摘要:\n for i, chunk in enumerate(chunks): # 取每块的前100个字符作为摘要 preview chunk[content][:100] ... if len(chunk[content]) 100 else chunk[content] summary f{i1}. {preview}\n return summary def optimize_context_usage( self, messages: List[Dict], current_token_count: int ) - List[Dict]: 优化上下文使用当接近token限制时进行压缩 if current_token_count self.max_tokens * 0.8: # 80%阈值 return messages # 压缩策略保留最重要的消息 compressed_messages [] # 总是保留系统消息和最近的消息 system_messages [msg for msg in messages if msg[role] system] recent_messages messages[-5:] # 保留最近5条消息 compressed_messages.extend(system_messages) # 对历史消息进行摘要 historical_messages messages[len(system_messages):-5] if historical_messages: summary self._create_conversation_summary(historical_messages) compressed_messages.append({ role: system, content: f先前对话摘要: {summary} }) compressed_messages.extend(recent_messages) return compressed_messages def _create_conversation_summary(self, messages: List[Dict]) - str: 创建对话摘要 user_messages [msg[content] for msg in messages if msg[role] user] assistant_messages [msg[content] for msg in messages if msg[role] assistant] summary f用户讨论了{len(user_messages)}个主题包括: {, .join(user_messages[:3])} if len(user_messages) 3: summary f 等{len(user_messages)}个话题 return summary5. Damon框架完整实战示例5.1 项目结构规划一个完整的Damon框架项目应该遵循清晰的结构组织damon-framework/ ├── src/ │ ├── damon/ │ │ ├── __init__.py │ │ ├── core/ │ │ │ ├── __init__.py │ │ │ ├── agent.py # 智能体核心类 │ │ │ ├── state_manager.py # 状态管理 │ │ │ └── dialogue_manager.py │ │ ├── tools/ │ │ │ ├── __init__.py │ │ │ ├── base.py # 工具基类 │ │ │ ├── web_tools.py # 网络工具 │ │ │ ├── file_tools.py # 文件工具 │ │ │ └── data_tools.py # 数据处理工具 │ │ ├── clients/ │ │ │ ├── __init__.py │ │ │ └── claude_client.py │ │ └── utils/ │ │ ├── __init__.py │ │ ├── config.py │ │ └── logger.py │ ├── tests/ │ │ ├── __init__.py │ │ ├── test_agent.py │ │ └── test_tools.py │ └── examples/ │ ├── simple_agent.py │ └── advanced_agent.py ├── docs/ │ ├── getting_started.md │ └── api_reference.md ├── requirements.txt ├── setup.py └── README.md5.2 基础智能体实现下面是一个完整的基础智能体实现示例# src/damon/core/agent.py import asyncio from typing import Dict, Any, Optional from ..clients.claude_client import ClaudeFable5Client from .dialogue_manager import DialogueManager from .state_manager import StateManager from ..tools.base import ToolRegistry class DamonAgent: def __init__(self, agent_id: str, config: Dict[str, Any] None): self.agent_id agent_id self.config config or {} self.tool_registry ToolRegistry() self.state_manager StateManager() self.dialogue_manager DialogueManager(self.tool_registry) self.is_running False # 注册内置工具 self._register_builtin_tools() def _register_builtin_tools(self): 注册内置工具 self.tool_registry.register_tool( nameget_current_time, description获取当前时间 ) async def get_current_time() - str: from datetime import datetime return datetime.now().isoformat() self.tool_registry.register_tool( namecreate_note, description创建笔记 ) async def create_note(title: str, content: str) - Dict[str, str]: return { id: fnote_{int(datetime.now().timestamp())}, title: title, content: content, created_at: datetime.now().isoformat() } async def start(self): 启动智能体 self.is_running True agent_state self.state_manager.get_agent_state(self.agent_id) agent_state.add_task_event(agent_started, {config: self.config}) print(f智能体 {self.agent_id} 已启动) async def process_query(self, query: str) - Dict[str, Any]: 处理用户查询 if not self.is_running: raise RuntimeError(智能体未启动) agent_state self.state_manager.get_agent_state(self.agent_id) agent_state.current_task f处理查询: {query[:50]}... agent_state.add_task_event(query_received, {query: query}) try: # 处理对话 response await self.dialogue_manager.process_message( query, agent_state.context ) # 更新状态 agent_state.context.update(response.get(updated_context, {})) agent_state.add_task_event(query_processed, { response: response.get(final_response, response.get(content, )), tools_used: [call[function][name] for call in response.get(tool_calls, [])] }) return response except Exception as e: agent_state.add_task_event(error_occurred, {error: str(e)}) raise async def stop(self): 停止智能体 self.is_running False agent_state self.state_manager.get_agent_state(self.agent_id) agent_state.add_task_event(agent_stopped, {}) # 保存状态到文件 self.state_manager.save_state(self.agent_id, f{self.agent_id}_state.json) print(f智能体 {self.agent_id} 已停止) # 使用示例 async def main(): # 创建智能体实例 agent DamonAgent(research-assistant, { max_context_length: 100000, allowed_tools: [web_search, create_note] }) # 启动智能体 await agent.start() try: # 处理查询 response await agent.process_query( 请搜索关于机器学习的最新进展并创建一份总结笔记 ) print(智能体响应:, response.get(final_response, response.get(content, ))) finally: # 停止智能体 await agent.stop() if __name__ __main__: asyncio.run(main())5.3 高级功能扩展对于更复杂的应用场景我们可以扩展基础智能体的功能# src/damon/core/advanced_agent.py from .agent import DamonAgent from typing import List, Dict, Any import asyncio class AdvancedDamonAgent(DamonAgent): def __init__(self, agent_id: str, config: Dict[str, Any] None): super().__init__(agent_id, config) self.task_queue asyncio.Queue() self.background_tasks set() async def schedule_task(self, task_description: str, priority: int 1): 调度后台任务 task { description: task_description, priority: priority, created_at: asyncio.get_event_loop().time() } await self.task_queue.put(task) async def _task_worker(self): 后台任务工作器 while self.is_running: try: task await asyncio.wait_for(self.task_queue.get(), timeout1.0) # 处理任务 result await self._process_background_task(task) # 记录任务完成 agent_state self.state_manager.get_agent_state(self.agent_id) agent_state.add_task_event(background_task_completed, { task: task, result: result }) self.task_queue.task_done() except asyncio.TimeoutError: continue async def _process_background_task(self, task: Dict) - Dict[str, Any]: 处理后台任务 # 这里可以实现复杂的任务处理逻辑 # 例如定期数据同步、监控告警等 response await self.process_query(task[description]) return { task_id: hash(task[description]), response: response, processed_at: asyncio.get_event_loop().time() } async def start(self): 启动高级智能体 await super().start() # 启动后台任务工作器 worker_task asyncio.create_task(self._task_worker()) self.background_tasks.add(worker_task) worker_task.add_done_callback(self.background_tasks.discard) async def stop(self): 停止高级智能体 await super().stop() # 取消所有后台任务 for task in self.background_tasks: task.cancel() await asyncio.gather(*self.background_tasks, return_exceptionsTrue)6. 生产环境部署与优化6.1 容器化部署配置对于生产环境推荐使用Docker进行容器化部署# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ ./src/ COPY examples/ ./examples/ # 创建非root用户 RUN useradd --create-home --shell /bin/bash damon USER damon # 设置环境变量 ENV PYTHONPATH/app/src ENV OPENROUTER_API_KEYyour_api_key_here # 启动命令 CMD [python, examples/advanced_agent.py]对应的Docker Compose配置# docker-compose.yml version: 3.8 services: damon-agent: build: . environment: - OPENROUTER_API_KEY${OPENROUTER_API_KEY} - LOG_LEVELINFO volumes: - ./data:/app/data restart: unless-stopped healthcheck: test: [CMD, python, -c, import requests; requests.get(http://localhost:8000/health)] interval: 30s timeout: 10s retries: 3 # 可选的监控服务 prometheus: image: prom/prometheus ports: - 9090:9090 volumes: - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml grafana: image: grafana/grafana ports: - 3000:3000 environment: - GF_SECURITY_ADMIN_PASSWORDadmin6.2 性能监控与日志管理生产环境需要完善的监控和日志系统# src/damon/utils/logger.py import logging import json from datetime import datetime from typing import Dict, Any class JSONFormatter(logging.Formatter): def format(self, record: logging.LogRecord) - str: log_entry { timestamp: datetime.utcnow().isoformat(), level: record.levelname, logger: record.name, message: record.getMessage(), module: record.module, function: record.funcName, line: record.lineno } if hasattr(record, extra_data): log_entry.update(record.extra_data) if record.exc_info: log_entry[exception] self.formatException(record.exc_info) return json.dumps(log_entry) def setup_logging(level: str INFO, log_file: str None): 设置结构化日志 logger logging.getLogger(damon) logger.setLevel(getattr(logging, level.upper())) # 避免重复添加handler if logger.handlers: return logger formatter JSONFormatter() # 控制台handler console_handler logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件handler如果指定了日志文件 if log_file: file_handler logging.FileHandler(log_file) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger # 性能监控装饰器 def monitor_performance(func): import time from functools import wraps wraps(func) async def wrapper(*args, **kwargs): start_time time.time() try: result await func(*args, **kwargs) duration time.time() - start_time # 记录性能指标 logger logging.getLogger(damon.performance) logger.info( fFunction {func.__name__} completed, extra{ extra_data: { function: func.__name__, duration: duration, status: success } } ) return result except Exception as e: duration time.time() - start_time logger logging.getLogger(damon.performance) logger.error( fFunction {func.__name__} failed, extra{ extra_data: { function: func.__name__, duration: duration, status: error, error: str(e) } } ) raise return wrapper6.3 安全最佳实践在生产环境中部署AI Agent需要特别注意安全性# src/damon/security/validator.py import re from typing import Any, Dict, List from urllib.parse import urlparse class SecurityValidator: def __init__(self): self.sensitive_patterns [ r\b(?:password|api[_-]?key|secret|token)\s*\s*[^\s], r\b(?:https?://[^\s]), # URL检测 ] self.blocked_domains [ internal.company.com, localhost, 127.0.0.1, 192.168.0.0/16, 10.0.0.0/8 ] def validate_input(self, user_input: str) - Dict[str, Any]: 验证用户输入的安全性 issues [] # 检查敏感信息 for pattern in self.sensitive_patterns: if re.search(pattern, user_input, re.IGNORECASE): issues.append(f检测到可能的敏感信息: {pattern}) # 检查URL安全性 urls re.findall
