企业级AI Agent实战:从知识冷启动到多智能体协作完整指南

📅 2026/7/14 23:50:59
企业级AI Agent实战:从知识冷启动到多智能体协作完整指南
如果你正在学习 AI Agent 开发可能会发现一个奇怪的现象网上的教程要么是简单的聊天机器人 Demo要么是复杂的学术论文真正能指导你从零搭建企业级智能体项目的实战内容少之又少。更让人困惑的是当你按照教程完成一个漂亮的 Demo 后却发现根本无法在实际业务中使用。这不是你的问题而是大多数教程只展示了冰山一角——他们告诉你如何让 AI 回答问题却没告诉你如何让 AI 在企业环境中可靠工作。根据行业数据超过 70% 的企业 AI Agent 项目在从 Demo 到生产环境的过程中失败。失败的原因不是模型能力不足而是企业级需求与演示级功能之间的巨大鸿沟。真正的企业级 AI Agent 需要处理复杂的文档格式、支持多轮对话、具备成本控制能力并满足安全合规要求。本文将基于汽车制造、酒店管理、医药零售等行业的真实落地案例手把手带你构建一个真正的企业级 AI Agent。我不会只给你另一个聊天机器人 Demo而是展示从知识冷启动、工作流编排到多智能体协作的完整生产路径。1. 这篇文章真正要解决的问题很多开发者对 AI Agent 存在误解认为它只是升级版的聊天机器人。这种误解导致他们在实际项目中踩坑无数。企业级 AI Agent 需要解决的是复杂业务场景下的动态决策问题而不仅仅是问答。举个例子传统的客服机器人只能回答预设的 FAQ 问题当用户提出我想把上周三的订单改到周五晚上并且把标准间换成套房这样的复杂请求时传统机器人就会失效。而真正的 AI Agent 能够理解这是修改订单请求识别需要更改的参数时间、房型并回退到相应的工作流节点执行操作。企业级 AI Agent 项目面临的核心挑战包括知识冷启动难题企业文档格式复杂PDF、Word、Excel、HTML包含表格、图片、嵌套结构简单的文本切分会导致上下文丢失意图识别局限用户不会用单轮命令说话需要全局上下文理解能力多智能体协作复杂度复杂业务需要多个专业化 Agent 协同工作而不是简单的提示词堆砌治理与运维盲区成本控制、安全合规、审计追溯等生产环境要求本文将重点解决这些实际问题带你避开 99% 的常见陷阱。2. AI Agent 基础概念与核心原理2.1 什么是真正的 AI AgentAI Agent智能体与聊天机器人的本质区别在于自主性。聊天机器人按照预设脚本响应而 AI Agent 能够根据目标自主规划步骤、调用工具、处理异常。通过一个对比表格可以更清晰理解维度聊天机器人工作流自动化AI AgentAgentic AI决策逻辑规则/意图匹配预定义流程LLM 驱动推理自主规划灵活性低脚本响应中分支逻辑高动态决策知识处理FAQ 查找结构化数据处理RAG 非结构化知识最适场景高频简单查询可重复业务流程复杂、上下文相关任务失败模式我不理解异常时流程中断幻觉、成本超支2.2 Agentic AI 的核心特征Agentic AI 具备以下关键能力自主规划根据目标分解任务步骤而不是执行固定脚本工具调用动态调用 API、数据库、外部服务异常处理在执行过程中遇到问题时能够自主调整策略上下文保持跨多轮对话维持状态和记忆持续学习从交互中优化策略和响应2.3 企业级 AI Agent 的典型架构一个生产级的 AI Agent 系统通常包含以下组件RAG 引擎负责知识检索支持多种文档格式和智能切分工作流编排器管理任务流程和状态转移意图识别模块理解用户真实意图和上下文工具调用层集成外部系统和 API记忆管理系统维护会话记忆和用户偏好治理控制台监控成本、安全、性能指标3. 环境准备与前置条件在开始构建之前我们需要准备好开发环境。企业级 AI Agent 项目对环境有特定要求不同于简单的 Demo 项目。3.1 硬件与网络要求内存至少 16GB RAM推荐 32GB用于处理大型语言模型和向量数据库存储100GB 可用空间文档存储和向量索引网络稳定访问模型 API 的网络环境GPU可选但如果有本地模型推理需求推荐配备 GPU3.2 软件环境准备# 创建 Python 虚拟环境 python -m venv ai-agent-env source ai-agent-env/bin/activate # Linux/Mac # ai-agent-env\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.1.0 pip install langchain-community0.0.10 pip install openai1.3.0 pip install chromadb0.4.15 pip install fastapi0.104.1 pip install uvicorn0.24.0 pip install pydantic2.5.03.3 开发工具配置# 项目结构初始化 mkdir enterprise-ai-agent cd enterprise-ai-agent mkdir -p src/{agents,workflows,knowledge,utils} mkdir -p data/{documents,vector_store} mkdir -p tests/{unit,integration} mkdir config # 创建基础配置文件 touch config/settings.py touch config/constants.py touch requirements.txt touch main.py3.4 API 密钥与安全配置在企业环境中安全配置至关重要。创建配置文件时要注意避免硬编码敏感信息# config/settings.py import os from typing import Optional from pydantic_settings import BaseSettings class Settings(BaseSettings): # API 配置 openai_api_key: Optional[str] None openai_base_url: Optional[str] None # 向量数据库配置 chroma_host: str localhost chroma_port: int 8000 # 应用配置 debug: bool False log_level: str INFO # 安全配置 max_tokens_per_minute: int 100000 # 成本控制 allowed_domains: list [your-company.com] class Config: env_file .env settings Settings() # 从环境变量加载敏感信息 if not settings.openai_api_key: settings.openai_api_key os.getenv(OPENAI_API_KEY)创建环境变量文件# .env 文件添加到 .gitignore OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLyour_base_url_optional4. 知识冷启动构建企业级 RAG 系统知识冷启动是大多数 AI Agent 项目失败的第一道坎。企业文档的复杂性远超普通文本需要专门的处理流程。4.1 企业文档的典型挑战企业文档处理面临四大挑战格式碎片化PDF、Word、Excel、HTML 等多种格式混合切分灾难机械切分导致上下文丢失表格盲区结构化数据被转换为混乱文本规模限制平台对文件大小和数量的限制4.2 智能文档解析器实现# src/knowledge/document_parser.py import os from typing import List, Dict, Any from pathlib import Path import PyPDF2 from docx import Document import pandas as pd from langchain.schema import Document as LangchainDocument class EnterpriseDocumentParser: 企业级文档解析器支持多种格式 def __init__(self, chunk_size: int 1000, chunk_overlap: int 200): self.chunk_size chunk_size self.chunk_overlap chunk_overlap def parse_pdf(self, file_path: Path) - List[LangchainDocument]: 解析 PDF 文档保留表格和结构信息 documents [] try: with open(file_path, rb) as file: pdf_reader PyPDF2.PdfReader(file) for page_num, page in enumerate(pdf_reader.pages): text page.extract_text() # 提取表格数据简化示例 tables self._extract_pdf_tables(page) # 创建文档对象保留元数据 doc LangchainDocument( page_contenttext, metadata{ source: str(file_path), page: page_num 1, tables: tables, type: pdf } ) documents.append(doc) except Exception as e: print(f解析 PDF 失败 {file_path}: {e}) return documents def parse_word(self, file_path: Path) - List[LangchainDocument]: 解析 Word 文档保留标题层级 documents [] try: doc Document(file_path) full_text [] for paragraph in doc.paragraphs: # 识别标题层级 style paragraph.style.name if paragraph.style else Normal text paragraph.text.strip() if text: full_text.append(f{style}: {text}) content \n.join(full_text) doc LangchainDocument( page_contentcontent, metadata{ source: str(file_path), type: word, structure: with_styles } ) documents.append(doc) except Exception as e: print(f解析 Word 文档失败 {file_path}: {e}) return documents def parse_excel(self, file_path: Path) - List[LangchainDocument]: 解析 Excel 文档保留表格结构 documents [] try: # 读取所有 sheet excel_file pd.ExcelFile(file_path) for sheet_name in excel_file.sheet_names: df pd.read_excel(file_path, sheet_namesheet_name) # 将 DataFrame 转换为结构化描述 table_description self._dataframe_to_description(df, sheet_name) doc LangchainDocument( page_contenttable_description, metadata{ source: str(file_path), sheet: sheet_name, type: excel, rows: len(df), columns: list(df.columns) } ) documents.append(doc) except Exception as e: print(f解析 Excel 失败 {file_path}: {e}) return documents def _extract_pdf_tables(self, page) - List[Dict]: 提取 PDF 中的表格数据简化实现 # 实际项目中可使用专门的 PDF 表格提取库 return [] def _dataframe_to_description(self, df: pd.DataFrame, sheet_name: str) - str: 将 DataFrame 转换为自然语言描述 description f工作表 {sheet_name} 包含以下数据\n\n description f表格有 {len(df)} 行{len(df.columns)} 列\n description f列名{, .join(df.columns)}\n\n # 添加前几行数据作为示例 if len(df) 0: description 前3行数据示例\n for i in range(min(3, len(df))): row_data {col: str(df.iloc[i][col]) for col in df.columns} description f第{i1}行: {row_data}\n return description4.3 智能文本切分策略传统按固定长度切分的方法会破坏文档的完整性我们需要更智能的切分策略# src/knowledge/text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter from typing import List import re class IntelligentTextSplitter: 智能文本切分器保留语义完整性 def __init__(self): # 使用递归切分优先按段落、句子、词语边界切分 self.splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap100, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) def split_documents(self, documents: List[LangchainDocument]) - List[LangchainDocument]: 切分文档保留上下文信息 split_docs [] for doc in documents: # 根据文档类型采用不同切分策略 if doc.metadata.get(type) excel: # 表格数据不切分保持完整 split_docs.append(doc) else: chunks self.splitter.split_text(doc.page_content) for i, chunk in enumerate(chunks): new_metadata doc.metadata.copy() new_metadata[chunk] i 1 chunk_doc LangchainDocument( page_contentchunk, metadatanew_metadata ) split_docs.append(chunk_doc) return split_docs def split_by_sections(self, text: str, headings: List[str]) - List[str]: 按标题切分文档 sections [] current_section current_heading lines text.split(\n) for line in lines: # 检测标题行 is_heading any(heading in line for heading in headings) if is_heading and current_section: # 保存当前章节 sections.append(f{current_heading}\n{current_section}) current_section current_heading line elif is_heading: current_heading line else: current_section line \n # 添加最后一个章节 if current_section: sections.append(f{current_heading}\n{current_section}) return sections4.4 向量数据库初始化与知识入库# src/knowledge/vector_store.py import chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings import os from typing import List class EnterpriseVectorStore: 企业级向量数据库管理 def __init__(self, persist_directory: str ./data/vector_store): self.persist_directory persist_directory self.embeddings OpenAIEmbeddings( modeltext-embedding-3-small, openai_api_keyos.getenv(OPENAI_API_KEY) ) # 创建 Chroma 客户端 self.client chromadb.PersistentClient(pathpersist_directory) # 创建或获取集合 self.collection self.client.get_or_create_collection( nameenterprise_documents, metadata{description: 企业知识库文档集合} ) self.vector_store Chroma( collection_nameenterprise_documents, embedding_functionself.embeddings, persist_directorypersist_directory ) def add_documents(self, documents: List[LangchainDocument]): 添加文档到向量数据库 try: # 分批处理避免内存溢出 batch_size 100 for i in range(0, len(documents), batch_size): batch documents[i:i batch_size] self.vector_store.add_documents(batch) print(f成功添加 {len(documents)} 个文档到向量数据库) except Exception as e: print(f添加文档失败: {e}) raise def similarity_search(self, query: str, k: int 5, **kwargs) - List[LangchainDocument]: 相似性搜索 return self.vector_store.similarity_search(query, kk, **kwargs) def get_collection_stats(self) - dict: 获取集合统计信息 return self.collection.count()5. 工作流编排构建智能业务逻辑知识库就绪后下一步是构建能够理解业务上下文的工作流系统。5.1 工作流引擎核心设计# src/workflows/workflow_engine.py from typing import Dict, Any, List, Optional from enum import Enum import json class NodeType(Enum): 工作流节点类型 PARAM_EXTRACTOR param_extractor # 参数提取 INTENT_CLASSIFIER intent_classifier # 意图分类 KNOWLEDGE_RETRIEVAL knowledge_retrieval # 知识检索 CODE_EXECUTION code_execution # 代码执行 CONDITIONAL_BRANCH conditional_branch # 条件分支 class WorkflowNode: 工作流节点基类 def __init__(self, node_id: str, node_type: NodeType, config: Dict[str, Any]): self.node_id node_id self.node_type node_type self.config config self.next_nodes [] # 后续节点 def execute(self, context: Dict[str, Any]) - Dict[str, Any]: 执行节点逻辑 raise NotImplementedError class ParamExtractorNode(WorkflowNode): 参数提取节点 def execute(self, context: Dict[str, Any]) - Dict[str, Any]: user_input context.get(user_input, ) current_params context.get(extracted_params, {}) # 使用 LLM 提取参数 extracted_params self._extract_parameters(user_input, current_params) # 更新上下文 context[extracted_params] {**current_params, **extracted_params} context[current_node] self.node_id return context def _extract_parameters(self, text: str, existing_params: Dict) - Dict: 从文本中提取参数 # 简化实现实际项目中可使用更复杂的 NLP 技术 import re params {} # 提取日期 date_pattern r(\d{4}年\d{1,2}月\d{1,2}日|明天|后天|下周[一二三四五六日]) dates re.findall(date_pattern, text) if dates: params[date] dates[0] # 提取数量 quantity_pattern r(\d)(个|位|人|间) quantities re.findall(quantity_pattern, text) if quantities: params[quantity] quantities[0][0] return params class IntentClassifierNode(WorkflowNode): 意图分类节点 def execute(self, context: Dict[str, Any]) - Dict[str, Any]: from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage user_input context.get(user_input, ) chat_history context.get(chat_history, []) # 构建提示词进行意图分类 prompt f 根据以下对话历史和当前用户输入判断用户意图 对话历史 {chat_history} 当前输入{user_input} 可选意图 - 查询信息用户想要获取特定信息 - 修改订单用户想要修改已有订单 - 新建请求用户发起新请求 - 投诉建议用户提出投诉或建议 请只返回意图名称。 llm ChatOpenAI(modelgpt-3.5-turbo) response llm([HumanMessage(contentprompt)]) intent response.content.strip() context[detected_intent] intent context[current_node] self.node_id return context class WorkflowEngine: 工作流引擎 def __init__(self): self.nodes {} self.start_node None def add_node(self, node: WorkflowNode): 添加节点 self.nodes[node.node_id] node def connect_nodes(self, from_node_id: str, to_node_id: str): 连接节点 if from_node_id in self.nodes and to_node_id in self.nodes: self.nodes[from_node_id].next_nodes.append(to_node_id) def execute_workflow(self, user_input: str, chat_history: List None) - Dict[str, Any]: 执行工作流 if chat_history is None: chat_history [] context { user_input: user_input, chat_history: chat_history, extracted_params: {}, current_node: None, workflow_result: None } current_node_id self.start_node execution_path [] while current_node_id: node self.nodes[current_node_id] execution_path.append(current_node_id) # 执行当前节点 context node.execute(context) # 决定下一个节点 current_node_id self._get_next_node(node, context) # 防止无限循环 if len(execution_path) 20: break context[execution_path] execution_path return context def _get_next_node(self, current_node: WorkflowNode, context: Dict) - Optional[str]: 决定下一个节点 if not current_node.next_nodes: return None # 简化实现选择第一个后续节点 # 实际项目中可根据条件判断选择不同分支 return current_node.next_nodes[0]5.2 全局意图识别实现企业级 AI Agent 需要理解整个对话上下文而不仅仅是当前消息# src/agents/intent_recognition.py from typing import Dict, List, Any import json class GlobalIntentRecognizer: 全局意图识别器 def __init__(self): self.conversation_context {} def analyze_conversation(self, user_id: str, current_input: str, chat_history: List[Dict]) - Dict[str, Any]: 分析整个对话的意图 # 构建完整的对话上下文 full_context self._build_context(chat_history, current_input) # 检测意图变化 intent_analysis self._detect_intent_shift(chat_history, current_input) # 识别参数引用如把时间改成7点中的时间指向之前的参数 parameter_references self._extract_parameter_references(chat_history, current_input) return { current_intent: intent_analysis.get(current_intent), intent_shift: intent_analysis.get(shift_detected, False), parameter_references: parameter_references, requires_clarification: self._needs_clarification(current_input, full_context), context_summary: self._summarize_context(full_context) } def _build_context(self, history: List[Dict], current_input: str) - str: 构建对话上下文 context_lines [] for i, turn in enumerate(history): role 用户 if i % 2 0 else 助手 context_lines.append(f{role}: {turn.get(content, )}) context_lines.append(f用户: {current_input}) return \n.join(context_lines) def _detect_intent_shift(self, history: List[Dict], current_input: str) - Dict: 检测意图是否发生变化 if len(history) 2: return {current_intent: new, shift_detected: False} # 分析最近几轮对话的意图连续性 recent_turns history[-4:] # 最近2轮对话 previous_intent self._classify_intent(recent_turns[0].get(content, )) if recent_turns else unknown current_intent self._classify_intent(current_input) shift_detected previous_intent ! current_intent return { previous_intent: previous_intent, current_intent: current_intent, shift_detected: shift_detected } def _classify_intent(self, text: str) - str: 简单意图分类 text_lower text.lower() if any(word in text_lower for word in [修改, 改成, 改变, 调整]): return modify elif any(word in text_lower for word in [查询, 查看, 了解, 想知道]): return query elif any(word in text_lower for word in [预订, 预约, 下单]): return book else: return unknown def _extract_parameter_references(self, history: List[Dict], current_input: str) - List[Dict]: 提取参数引用 references [] # 查找当前输入中提到的参数名 parameter_keywords [时间, 日期, 人数, 房型, 订单] for keyword in parameter_keywords: if keyword in current_input: # 在历史中查找该参数之前的值 previous_value self._find_parameter_in_history(history, keyword) if previous_value: references.append({ parameter: keyword, previous_value: previous_value, current_mention: current_input }) return references def _find_parameter_in_history(self, history: List[Dict], parameter: str) - str: 在历史对话中查找参数值 for turn in reversed(history): content turn.get(content, ) if parameter in content: # 简化实现实际可使用更精确的提取 return content return 6. 多智能体协作架构对于复杂企业场景单个 Agent 无法满足所有需求需要多智能体协作。6.1 智能体协作模式设计# src/agents/multi_agent_system.py from typing import Dict, List, Any, Optional from abc import ABC, abstractmethod import uuid from datetime import datetime class Agent(ABC): 智能体基类 def __init__(self, agent_id: str, name: str, capabilities: List[str]): self.agent_id agent_id self.name name self.capabilities capabilities self.conversation_memory {} abstractmethod def process_request(self, request: Dict) - Dict: 处理请求 pass def can_handle(self, intent: str, context: Dict) - bool: 判断是否能处理该意图 return intent in self.capabilities class CustomerServiceAgent(Agent): 客服智能体 def __init__(self): super().__init__( agent_idcs_agent_001, name客户服务专家, capabilities[query, complaint, general_help] ) def process_request(self, request: Dict) - Dict: user_input request.get(user_input, ) context request.get(context, {}) # 基于知识库提供回答 response self._generate_response(user_input, context) return { agent_id: self.agent_id, response: response, confidence: 0.85, suggested_next_agent: None # 可以自行处理不需要转交 } def _generate_response(self, user_input: str, context: Dict) - str: 生成响应 # 简化实现实际可集成 RAG 系统 return f您好关于\{user_input}\的问题我查询到相关信息如下[...] class BookingAgent(Agent): 预订智能体 def __init__(self): super().__init__( agent_idbooking_agent_001, name预订专家, capabilities[book, modify, cancel] ) def process_request(self, request: Dict) - Dict: user_input request.get(user_input, ) context request.get(context, {}) # 分析是新建预订还是修改现有预订 intent self._analyze_booking_intent(user_input, context) if intent modify: # 检查是否有足够的参数进行修改 if self._has_sufficient_modification_params(context): response self._process_modification(user_input, context) next_agent None else: response 请问您想要修改订单的哪些信息 next_agent None else: response self._process_new_booking(user_input, context) next_agent None return { agent_id: self.agent_id, response: response, confidence: 0.92, suggested_next_agent: next_agent } class MultiAgentCoordinator: 多智能体协调器 def __init__(self): self.agents {} self.register_agents() def register_agents(self): 注册智能体 self.agents[customer_service] CustomerServiceAgent() self.agents[booking] BookingAgent() def route_request(self, user_input: str, context: Dict) - Dict: 路由请求到合适的智能体 # 第一步意图识别 intent self._classify_intent(user_input, context) # 第二步选择最合适的智能体 selected_agent self._select_agent(intent, context) if selected_agent: # 第三步执行处理 request_data { user_input: user_input, context: context, intent: intent } result selected_agent.process_request(request_data) # 第四步处理转交逻辑 if result.get(suggested_next_agent): next_agent self.agents.get(result[suggested_next_agent]) if next_agent: # 将当前结果作为上下文传递给下一个智能体 new_context {**context, previous_agent_result: result} final_result next_agent.process_request({ user_input: user_input, context: new_context }) return final_result return result else: return { response: 抱歉我暂时无法处理这个请求。, confidence: 0.0, suggested_next_agent: customer_service } def _classify_intent(self, user_input: str, context: Dict) - str: 意图分类 # 简化实现实际可使用更复杂的 NLP 模型 user_input_lower user_input.lower() if any(word in user_input_lower for word in [预订, 预约, 下单]): return book elif any(word in user_input_lower for word in [修改, 改成, 改变]): return modify elif any(word in user_input_lower for word in [投诉, 建议, 不满意]): return complaint else: return query def _select_agent(self, intent: str, context: Dict) - Optional[Agent]: 选择智能体 suitable_agents [] for agent in self.agents.values(): if agent.can_handle(intent, context): suitable_agents.append(agent) if suitable_agents: # 选择第一个合适的智能体可扩展为基于置信度选择 return suitable_agents[0] return None7. 企业级治理与运维生产环境中的 AI Agent 需要完善的治理机制。7.1 成本控制与监控# src/utils/cost_monitor.py from typing import Dict, List import time from datetime import datetime, timedelta import json class CostMonitor: 成本监控器 def __init__(self, budget_limits: Dict[str, float]): self.budget_limits budget_limits # 例如: {daily: 100.0, monthly: 3000.0} self.usage_data { daily: {tokens: 0, cost: 0.0, date: datetime.now().date()}, monthly: {tokens: 0, cost: 0.0, month: datetime.now().month} } self.alert_threshold 0.8 # 80% 预算时告警 def record_usage(self, tokens: int, cost: float, operation: str): 记录使用情况 current_date datetime.now().date() current_month datetime.now().month # 检查是否需要重置计数器 if self.usage_data[daily][date] ! current_date: self.usage_data[daily] {tokens: 0, cost: 0.0, date: current_date} if self.usage_data[monthly][month] ! current_month: self.usage_data[monthly] {tokens: 0, cost: 0.0, month: current_month} # 更新使用数据 self.usage_data[daily][tokens] tokens self.usage_data[daily][cost] cost self.usage_data[monthly][tokens] tokens self.usage_data[monthly][cost] cost # 检查预算限制 self._check_budget_limits() def _check_budget_limits(self): 检查预算限制 alerts [] for period, limit in self.budget_limits.items(): current_usage self.usage_data[period][cost] usage_ratio current_usage / limit if limit 0 else 0 if usage_ratio 1.0: alerts.append(f{period}预算超支当前使用: {current_usage}, 限制: {limit}) elif usage_ratio self.alert_threshold: alerts.append(f{period}预算使用超过{self.alert_threshold*100}%: {current_usage}/{limit}) if alerts: self._send_alerts(alerts) def _send_alerts(self, alerts: List[str]): 发送告警 for alert in alerts: print(f成本告警: {alert}) # 实际项目中可集成邮件、短信、钉钉等告警方式 def get_usage_report(self) - Dict: 获取使用报告 return { daily_usage: self.usage_data[daily], monthly_usage: self.usage_data[monthly], budget_limits: self.budget_limits, timestamp: datetime.now().isoformat() }7.2 安全与合规控制# src/utils/security_controller.py import re from typing import Dict, List, Optional class SecurityController: 安全控制器 def __init__(self): self.sensitive_patterns [ r\b\d{15,19}\b, # 银行卡号 r\b\d{17}[\dXx]\b, #