在企业级AI应用开发中如何将多个智能体高效协同工作一直是个技术难点。最近在业务项目中尝试LangChainLangGraphMCP组合方案时发现现有资料要么过于零散要么缺乏实战深度。本文整合一套完整的企业级开发指南从基础概念到多智能体工作流实战包含可运行的代码示例和部署避坑方案适合有一定Python基础的开发者快速上手。1. 智能体开发技术栈核心概念解析1.1 LangChainAI应用开发的基础框架LangChain是一个用于开发基于语言模型应用的框架它提供了统一的接口和组件让开发者能够快速构建复杂的AI应用。在企业级开发中LangChain主要解决以下问题模块化设计将AI应用拆分为可重用的组件如提示模板、链式调用、记忆存储等工具集成支持外部API、数据库、文件系统等工具的标准化接入可扩展性允许自定义组件和工具适应企业特定需求# 基础LangChain使用示例 from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain.llms import OpenAI # 创建提示模板 prompt PromptTemplate( input_variables[product], template为{product}写一个吸引人的广告标语 ) # 初始化语言模型 llm OpenAI(temperature0.7) # 创建链 chain LLMChain(llmllm, promptprompt) # 执行链 result chain.run(智能手表) print(result)1.2 LangGraph多智能体工作流编排引擎LangGraph建立在LangChain之上专门用于构建有状态的多步骤工作流。与LangChain的链式调用不同LangGraph支持图结构将AI应用建模为有向图节点代表处理步骤边代表控制流状态管理在整个工作流中维护和更新共享状态条件分支根据中间结果动态选择执行路径循环控制支持迭代执行直到满足特定条件from langgraph.graph import Graph from langgraph.checkpoint.sqlite import SqliteSaver # 创建基础图结构 graph Graph() # 定义节点函数 def node1(state): return {step1_result: f处理完成: {state[input]}} def node2(state): return {step2_result: f进一步处理: {state[step1_result]}} # 添加节点和边 graph.add_node(node1, node1) graph.add_node(node2, node2) graph.add_edge(node1, node2) # 设置入口点 graph.set_entry_point(node1)1.3 MCPModel Context Protocol模型上下文协议MCP是一种新兴的协议标准旨在解决AI应用中的上下文管理问题。在企业级应用中MCP提供统一上下文管理标准化不同组件间的上下文传递格式工具发现机制动态发现和注册可用工具安全边界控制工具访问权限确保企业数据安全跨平台兼容支持不同AI模型和框架的互操作1.4 Agent智能体自主决策的AI单元智能体是能够感知环境、做出决策并执行动作的AI实体。在企业级开发中智能体通常具备工具使用能力调用外部API和系统功能记忆机制保留对话历史和执行上下文推理能力基于当前状态制定行动计划协作能力与其他智能体协同完成复杂任务2. 环境准备与版本兼容性配置2.1 开发环境要求企业级AI应用开发需要稳定的环境支持建议配置如下操作系统Ubuntu 20.04 / Windows 10 / macOS 12Python版本3.8-3.11推荐3.9内存要求至少8GB推荐16GB以上网络环境稳定的互联网连接用于API调用2.2 核心依赖安装与版本管理正确的版本搭配是项目成功的关键以下是经过验证的稳定版本组合# 创建虚拟环境 python -m venv langchain_env source langchain_env/bin/activate # Linux/macOS # 或 langchain_env\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.1.0 pip install langgraph0.0.40 pip install langchain-openai0.0.8 pip install langchain-community0.0.20 # 可选数据库支持 pip install sqlite3 # 内置通常不需要单独安装 pip install pymysql1.1.0 # MySQL连接2.3 开发工具配置为提高开发效率建议配置以下工具# requirements.txt 示例 langchain0.1.0 langgraph0.0.40 langchain-openai0.0.8 langchain-community0.0.20 openai1.3.0 pydantic2.5.0 sqlite3 python-dotenv1.0.0 # 环境变量管理 # .env.example 环境变量模板 OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLsqlite:///./app.db LOG_LEVELINFO3. LangChain核心组件深度解析3.1 提示模板与链式调用提示模板是LangChain的基础构建块支持变量插值和动态内容生成from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate from langchain.schema import SystemMessage # 复杂提示模板示例 system_template SystemMessage(content你是一个专业的产品营销专家。) human_template HumanMessagePromptTemplate.from_template( 为{product}设计一个营销方案目标客户是{audience}预算为{budget}元。 ) chat_prompt ChatPromptTemplate.from_messages([system_template, human_template]) # 格式化提示 formatted_prompt chat_prompt.format_prompt( product智能家居系统, audience年轻家庭, budget50000 ) print(formatted_prompt.to_string())3.2 记忆管理机制记忆管理是智能体应用的核心LangChain提供多种记忆存储方案from langchain.memory import ConversationBufferMemory, SQLiteMemory # 对话记忆管理 memory ConversationBufferMemory( return_messagesTrue, memory_keychat_history, output_keyoutput ) # 数据库记忆存储 sql_memory SQLiteMemory( database_urlsqlite:///./memory.db, memory_keychat_history ) # 记忆使用示例 def chat_with_memory(user_input, memory_instance): # 从记忆加载历史 history memory_instance.load_memory_variables({}) # 构建当前对话上下文 context f历史对话{history}\n当前输入{user_input} # 处理并保存新记忆 memory_instance.save_context( {input: user_input}, {output: f已处理{user_input}} ) return context3.3 工具集成与自定义工具开发工具是智能体与外部世界交互的桥梁from langchain.tools import BaseTool from typing import Type from pydantic import BaseModel, Field # 自定义工具输入模型 class CalculatorInput(BaseModel): a: float Field(description第一个数字) b: float Field(description第二个数字) operation: str Field(description运算类型add, subtract, multiply, divide) # 自定义计算器工具 class CalculatorTool(BaseTool): name calculator description 执行数学运算 args_schema: Type[BaseModel] CalculatorInput def _run(self, a: float, b: float, operation: str) - float: if operation add: return a b elif operation subtract: return a - b elif operation multiply: return a * b elif operation divide: if b 0: raise ValueError(除数不能为零) return a / b else: raise ValueError(不支持的运算类型) # 工具使用示例 calculator CalculatorTool() result calculator.run({a: 10, b: 5, operation: multiply}) print(f计算结果{result})4. LangGraph工作流设计与实现4.1 基础图结构构建LangGraph使用图结构来定义复杂的工作流from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator # 定义状态类型 class GraphState(TypedDict): input: str processed_data: Annotated[list, operator.add] decision: str final_result: str # 创建状态图 graph_builder StateGraph(GraphState) # 定义节点函数 def data_processing_node(state: GraphState) - GraphState: 数据处理节点 processed state[input].upper().split() return {processed_data: processed} def decision_node(state: GraphState) - GraphState: 决策节点 data_length len(state[processed_data]) decision complex if data_length 3 else simple return {decision: decision} def simple_processing_node(state: GraphState) - GraphState: 简单处理节点 result | .join(state[processed_data]) return {final_result: f简单处理: {result}} def complex_processing_node(state: GraphState) - GraphState: 复杂处理节点 result - .join(state[processed_data]) return {final_result: f复杂处理: {result}} # 添加节点 graph_builder.add_node(process, data_processing_node) graph_builder.add_node(decide, decision_node) graph_builder.add_node(simple, simple_processing_node) graph_builder.add_node(complex, complex_processing_node) # 设置边和条件路由 graph_builder.set_entry_point(process) graph_builder.add_edge(process, decide) # 条件路由 def route_decision(state: GraphState) - str: if state[decision] simple: return simple else: return complex graph_builder.add_conditional_edges( decide, route_decision, {simple: simple, complex: complex} ) graph_builder.add_edge(simple, END) graph_builder.add_edge(complex, END) # 编译图 graph graph_builder.compile()4.2 状态管理与检查点机制LangGraph的状态管理确保工作流的持久化和恢复能力from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 创建检查点存储 checkpointer SqliteSaver.from_conn(sqlite3.connect(checkpoints.db)) # 带检查点的图配置 graph_with_checkpoint graph_builder.compile( checkpointercheckpointer ) # 执行带状态管理的工作流 def execute_workflow_with_state(input_data, thread_iddefault): initial_state {input: input_data, processed_data: []} # 执行工作流 final_state graph_with_checkpoint.invoke( initial_state, {configurable: {thread_id: thread_id}} ) return final_state # 示例执行 result1 execute_workflow_with_state(hello world test example, thread_1) print(f第一次执行结果: {result1}) # 可以从检查点恢复执行 result2 graph_with_checkpoint.invoke( {input: additional input}, {configurable: {thread_id: thread_1}} ) print(f恢复执行结果: {result2})4.3 多智能体协作工作流在企业级应用中多个智能体需要协同工作from langgraph.graph import MessagesState from langchain_core.messages import HumanMessage, AIMessage class MultiAgentState(MessagesState): agent1_result: str None agent2_result: str None final_decision: str None def create_multi_agent_workflow(): 创建多智能体协作工作流 builder StateGraph(MultiAgentState) # 专业分析智能体 def analyst_agent(state: MultiAgentState): last_message state.messages[-1] analysis f分析结果: 输入内容{last_message.content}包含重要信息 return {agent1_result: analysis} # 决策智能体 def decision_agent(state: MultiAgentState): based_on state.agent1_result decision f基于分析{based_on}建议采取行动 return {agent2_result: decision} # 协调智能体 def coordinator_agent(state: MultiAgentState): final_output f协调结果: {state.agent1_result} {state.agent2_result} return {final_decision: final_output} # 构建工作流 builder.add_node(analyst, analyst_agent) builder.add_node(decision_maker, decision_agent) builder.add_node(coordinator, coordinator_agent) builder.set_entry_point(analyst) builder.add_edge(analyst, decision_maker) builder.add_edge(decision_maker, coordinator) builder.add_edge(coordinator, END) return builder.compile() # 使用多智能体工作流 multi_agent_graph create_multi_agent_workflow() initial_messages [HumanMessage(content需要分析的市场数据)] result multi_agent_graph.invoke({messages: initial_messages}) print(f多智能体协作结果: {result})5. MCP协议在企业级应用中的实践5.1 MCP服务器开发与集成MCP服务器作为工具和模型的桥梁需要标准化开发import asyncio from mcp import MCPServer from mcp.server.models import InitializationOptions from mcp.types import Tool, TextContent class EnterpriseMCPServer: def __init__(self): self.server MCPServer(enterprise-tools) async def initialize(self, options: InitializationOptions): 初始化MCP服务器 # 注册可用工具 tools [ Tool( namedata_analyzer, description企业数据分析工具, inputSchema{ type: object, properties: { dataset: {type: string}, analysis_type: {type: string} } } ), Tool( namereport_generator, description自动化报告生成, inputSchema{ type: object, properties: { data: {type: string}, format: {type: string} } } ) ] return tools async def handle_tool_call(self, name: str, arguments: dict): 处理工具调用 if name data_analyzer: return await self._analyze_data(arguments) elif name report_generator: return await self._generate_report(arguments) else: raise ValueError(f未知工具: {name}) async def _analyze_data(self, args: dict): 数据分析实现 dataset args.get(dataset, ) analysis_type args.get(analysis_type, basic) # 模拟数据分析逻辑 analysis_result f对数据集{dataset}执行{analysis_type}分析完成 return [TextContent(typetext, textanalysis_result)] async def _generate_report(self, args: dict): 报告生成实现 data args.get(data, ) report_format args.get(format, summary) report f生成{report_format}格式报告基于数据: {data} return [TextContent(typetext, textreport)] # 启动MCP服务器 async def main(): server EnterpriseMCPServer() # 这里简化了实际启动逻辑 print(MCP服务器初始化完成) # asyncio.run(main())5.2 工具发现与动态注册MCP支持运行时工具发现增强系统灵活性class DynamicToolManager: def __init__(self): self.registered_tools {} def register_tool(self, name: str, description: str, function: callable): 动态注册工具 self.registered_tools[name] { description: description, function: function } def discover_tools(self): 发现可用工具 return [ { name: name, description: info[description], parameters: self._inspect_function_parameters(info[function]) } for name, info in self.registered_tools.items() ] def execute_tool(self, name: str, **kwargs): 执行工具 if name not in self.registered_tools: raise ValueError(f工具未注册: {name}) tool_function self.registered_tools[name][function] return tool_function(**kwargs) def _inspect_function_parameters(self, func: callable): 检查函数参数简化实现 import inspect sig inspect.signature(func) return { name: str(param.annotation) for name, param in sig.parameters.items() } # 使用示例 tool_manager DynamicToolManager() # 动态注册工具 def custom_calculator(x: float, y: float) - float: return x * y 100 tool_manager.register_tool( business_calculator, 企业级业务计算器, custom_calculator ) # 发现和使用工具 available_tools tool_manager.discover_tools() print(f可用工具: {available_tools}) result tool_manager.execute_tool(business_calculator, x10, y5) print(f工具执行结果: {result})6. 企业级智能体开发实战6.1 需求分析与系统设计假设我们需要开发一个企业智能客服系统需求如下多轮对话支持上下文感知的连续对话工具集成能够查询订单、产品信息等业务数据工作流管理复杂问题自动路由到相应处理流程状态持久化对话状态支持暂停和恢复系统架构设计用户输入 → 路由智能体 → 工具调用 → 业务处理 → 响应生成 ↓ 状态管理 ← 记忆存储6.2 核心智能体实现from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI class EnterpriseCustomerServiceAgent: def __init__(self, tools, system_promptNone): self.tools tools self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.3) self.prompt self._create_agent_prompt(system_prompt) self.agent self._create_agent() self.agent_executor AgentExecutor( agentself.agent, toolstools, verboseTrue, handle_parsing_errorsTrue ) def _create_agent_prompt(self, system_prompt): 创建智能体提示模板 default_system 你是企业级智能客服助手具有以下能力 1. 处理客户咨询和问题解答 2. 使用工具查询业务信息 3. 多轮对话保持上下文连贯 4. 复杂问题自动转接专业处理 请专业、友好地帮助客户解决问题。 system_content system_prompt or default_system prompt ChatPromptTemplate.from_messages([ (system, system_content), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}) ]) return prompt def _create_agent(self): 创建智能体 return create_tool_calling_agent(self.llm, self.tools, self.prompt) def process_query(self, user_input, chat_historyNone): 处理用户查询 input_data { input: user_input, chat_history: chat_history or [] } try: response self.agent_executor.invoke(input_data) return response[output] except Exception as e: return f处理请求时出现错误: {str(e)} # 业务工具定义 def query_order_status(order_id: str) - str: 查询订单状态模拟实现 # 实际项目中这里会连接数据库或API return f订单 {order_id} 状态已发货预计明天送达 def get_product_info(product_id: str) - str: 获取产品信息模拟实现 return f产品 {product_id} 信息智能家居套装价格2999元 def escalate_to_human_agent(issue: str) - str: 转接人工客服 return f问题已转接人工客服{issue} # 创建智能体实例 from langchain.tools import tool tool def order_status_tool(order_id: str) - str: 查询订单状态工具 return query_order_status(order_id) tool def product_info_tool(product_id: str) - str: 查询产品信息工具 return get_product_info(product_id) tool def escalate_tool(issue: str) - str: 转接人工客服工具 return escalate_to_human_agent(issue) tools [order_status_tool, product_info_tool, escalate_tool] customer_agent EnterpriseCustomerServiceAgent(tools) # 测试智能体 response customer_agent.process_query(我的订单12345状态如何) print(f智能体响应: {response})6.3 多智能体协同系统复杂业务场景需要多个智能体协同工作class MultiAgentCoordinator: def __init__(self): self.agents {} self.workflow_graph None def register_agent(self, name: str, agent_instance, capabilities: list): 注册智能体 self.agents[name] { instance: agent_instance, capabilities: capabilities } def create_coordination_workflow(self): 创建智能体协调工作流 from langgraph.graph import StateGraph from typing import TypedDict class CoordinationState(TypedDict): user_input: str current_agent: str agent_responses: dict final_output: str builder StateGraph(CoordinationState) def input_router(state: CoordinationState): 输入路由到合适智能体 user_input state[user_input] # 基于关键词的简单路由逻辑 if 订单 in user_input: return {current_agent: order_agent} elif 产品 in user_input: return {current_agent: product_agent} else: return {current_agent: general_agent} def execute_agent(state: CoordinationState): 执行智能体处理 agent_name state[current_agent] user_input state[user_input] if agent_name not in self.agents: return {final_output: 未找到合适的处理智能体} agent self.agents[agent_name][instance] response agent.process_query(user_input) responses state.get(agent_responses, {}) responses[agent_name] response return { agent_responses: responses, final_output: response } # 构建工作流 builder.add_node(router, input_router) builder.add_node(executor, execute_agent) builder.set_entry_point(router) builder.add_edge(router, executor) builder.add_edge(executor, END) self.workflow_graph builder.compile() return self.workflow_graph def process_user_request(self, user_input: str): 处理用户请求 if not self.workflow_graph: self.create_coordination_workflow() initial_state {user_input: user_input} result self.workflow_graph.invoke(initial_state) return result[final_output] # 使用多智能体协调系统 coordinator MultiAgentCoordinator() # 注册不同专业领域的智能体 general_agent EnterpriseCustomerServiceAgent(tools) order_agent EnterpriseCustomerServiceAgent([order_status_tool, escalate_tool]) product_agent EnterpriseCustomerServiceAgent([product_info_tool, escalate_tool]) coordinator.register_agent(general_agent, general_agent, [通用咨询]) coordinator.register_agent(order_agent, order_agent, [订单查询]) coordinator.register_agent(product_agent, product_agent, [产品信息]) # 测试协同处理 test_queries [ 我想了解产品ABC的信息, 我的订单67890到哪里了, 你们公司的联系方式是什么 ] for query in test_queries: response coordinator.process_user_request(query) print(f用户问题: {query}) print(f系统响应: {response}\n)7. 项目部署与生产环境配置7.1 容器化部署方案使用Docker确保环境一致性# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ sqlite3 \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建数据目录 RUN mkdir -p /app/data # 设置环境变量 ENV PYTHONPATH/app ENV DATABASE_URLsqlite:////app/data/app.db # 启动应用 CMD [python, main.py]7.2 配置管理最佳实践生产环境配置需要安全性和灵活性# config.py import os from typing import Optional from pydantic import BaseSettings, Field class Settings(BaseSettings): 应用配置 # API配置 openai_api_key: str Field(..., envOPENAI_API_KEY) api_timeout: int Field(30, envAPI_TIMEOUT) # 数据库配置 database_url: str Field(sqlite:///./app.db, envDATABASE_URL) # 应用配置 log_level: str Field(INFO, envLOG_LEVEL) max_workers: int Field(4, envMAX_WORKERS) # 安全配置 allowed_origins: list Field([http://localhost:3000], envALLOWED_ORIGINS) class Config: env_file .env case_sensitive False # 配置实例 settings Settings() # 安全密钥管理 def get_api_key(service_name: str) - Optional[str]: 安全获取API密钥 try: if service_name openai: return settings.openai_api_key # 可以扩展其他服务 return None except Exception as e: print(f获取API密钥失败: {e}) return None7.3 监控与日志记录完善的监控体系是生产环境的必备条件import logging import json from datetime import datetime class JSONFormatter(logging.Formatter): JSON日志格式化器 def format(self, record): 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 record.exc_info: log_entry[exception] self.formatException(record.exc_info) return json.dumps(log_entry, ensure_asciiFalse) def setup_logging(): 配置日志系统 # 创建logger logger logging.getLogger(enterprise_ai) logger.setLevel(logging.INFO) # 避免重复添加handler if logger.handlers: return logger # 文件handler file_handler logging.FileHandler(app.log, encodingutf-8) file_handler.setFormatter(JSONFormatter()) # 控制台handler console_handler logging.StreamHandler() console_handler.setFormatter( logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) ) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 使用示例 logger setup_logging() def log_agent_interaction(user_input: str, agent_response: str, success: bool True): 记录智能体交互日志 log_data { user_input: user_input, agent_response: agent_response, success: success, timestamp: datetime.utcnow().isoformat() } if success: logger.info(智能体交互完成, extralog_data) else: logger.error(智能体交互失败, extralog_data)8. 常见问题与解决方案8.1 依赖版本冲突问题LangChain生态版本更新较快常见冲突及解决方案问题现象可能原因解决方案ImportError: cannot import name版本不兼容统一使用测试过的版本组合AttributeError: module has no attributeAPI变更查看对应版本的文档运行时类型错误Pydantic版本冲突固定pydantic2.5.0# 版本兼容性检查脚本 import pkg_resources def check_dependencies(): requirements { langchain: 0.1.0, langgraph: 0.0.40, pydantic: 2.5.0 } for package, expected_version in requirements.items(): try: installed_version pkg_resources.get_distribution(package).version if installed_version ! expected_version: print(f警告: {package} 版本不匹配。已安装: {installed_version}, 期望: {expected_version}) else: print(f✓ {package} 版本正确: {installed_version}) except pkg_resources.DistributionNotFound: print(f错误: {package} 未安装) check_dependencies()8.2 内存管理与性能优化智能体应用容易遇到内存问题优化策略包括import gc import psutil import threading from contextlib import contextmanager class MemoryMonitor: 内存监控器 def __init__(self, threshold_mb: int 1024): self.threshold threshold_mb self.monitoring False contextmanager def auto_cleanup(self): 自动内存清理上下文管理器 try: yield finally: self.force_garbage_collection() def force_garbage_collection(self): 强制垃圾回收 collected gc.collect() print(f垃圾回收完成释放对象: {collected}) def get_memory_usage(self): 获取内存使用情况 process psutil.Process() memory_info process.memory_info() return memory_info.rss / 1024 / 1024 # 转换为MB def check_memory_threshold(self): 检查内存阈值 current_usage self.get_memory_usage() if current_usage self.threshold: print(f内存使用过高: {current_usage:.2f}MB {self.threshold}MB) self.force_garbage_collection() return True return False # 使用内存监控 memory_monitor MemoryMonitor(threshold_mb512) def optimized_agent_processing(user_input): 带内存优化的智能体处理 with memory_monitor.auto_cleanup(): # 内存密集型操作 result customer_agent.process_query(user_input) # 检查内存使用 memory_monitor.check_memory_threshold() return result8.3 网络超时与重试机制企业级应用需要处理网络不稳定性import asyncio from typing import Type, Tuple from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class NetworkManager: 网络连接管理器 def __init__(self, max_retries: int 3, base_delay: float 1.0): self.max_retries max_retries self.base_delay base_delay retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type((TimeoutError, ConnectionError)) ) async def api_call_with_retry(self, api_function, *args, **kwargs): 带重试的API调用 try: response await api_function(*args, **kwargs) return response except (TimeoutError, ConnectionError) as e: print(fAPI调用失败: {e}进行重试...) raise def create_timeout_context(self, timeout_seconds: int 30): 创建超时上下文 return asyncio.timeout(timeout_seconds) # 使用示例 async def reliable_agent_call(user_input: str): 可靠的智能体调用 network_manager NetworkManager() async def agent_processing(): # 模拟智能体处理 await asyncio.sleep(2) return f处理结果: {user_input} try: async with network_manager.create_timeout_context(10): result await network_manager.api_call_with_retry(agent_processing) return result except asyncio.TimeoutError: return 请求超时请稍后重试 except Exception as e: return f处理失败: {str(e)} # 测试网络容错 # asyncio.run(reliable_agent_call(测试输入))9. 安全最佳实践9.1 输入验证与 sanitization防止注入攻击和恶意输入import re from html import escape from typing import Any, Dict class InputValidator: 输入验证器 def __init__(self): self.suspicious_patterns [ rscript.*?.*?/script, # 脚本标签 rjavascript:, # JavaScript协议 ron\w\s*, # 事件处理器 runion\sselect, # SQL注入 r;\s*(drop|delete|update