基于AWS Bedrock构建药物发现AI智能体：从概念到工程实践

在实际药物研发项目中，从靶点发现到候选化合物筛选，再到临床前研究，是一个漫长且成本高昂的过程。传统方法依赖大量实验试错，周期以年计。近年来，以生成式AI为代表的人工智能技术，正被探索用于加速这一进程，特别是在分子生成、性质预测和实验模拟等环节。大型制药公司与科技巨头的合作，例如诺和诺德与AWS的合作，其核心是将“智能体”这一更高级的AI范式引入药物发现流程，旨在构建能够自主规划、调用工具、并持续学习的AI系统，而不仅仅是执行单一预测任务。

这种合作模式揭示了一个关键趋势：药物发现的AI应用正从“单点工具”向“自动化工作流”演进。对于技术开发者而言，理解如何构建和集成这样的AI智能体系统，比单纯使用某个预测模型更具前瞻性。本文将围绕“AI智能体在药物发现中的应用”这一主线，拆解其核心概念、技术架构，并通过一个模拟化合物性质筛选的简化示例，展示如何利用AWS云服务搭建一个具备基础智能体能力的原型系统。通过本文，你将理解智能体与普通AI模型的区别，掌握使用AWS Bedrock等工具构建可执行多步任务的AI应用的基本方法，并了解在实际药物研发场景中集成此类系统需要考虑的关键工程问题。

1. 理解药物发现中的AI智能体：从预测工具到自主工作流

在讨论具体技术实现前，必须厘清“AI智能体”在药物发现语境下的特殊含义。它并非一个全新的算法，而是一种系统设计范式。

1.1 传统AI模型与AI智能体的核心区别

在早期的AI辅助药物发现中，模型通常扮演一个“静态预测器”的角色。例如，给定一个分子结构（SMILES字符串），一个训练好的模型可以预测其水溶性（LogP）或与特定靶点的结合亲和力（pIC50）。这个过程是单向的：输入分子，输出预测值。开发者需要事先明确任务、准备好标注数据、训练模型，并在应用时严格限定输入格式。

AI智能体则引入了“自主性”和“序列决策”能力。一个药物发现智能体可以被赋予一个高层目标，例如“找到一个对靶点X具有高活性且类药性好的分子”。为了达成这个目标，智能体需要自主规划一系列步骤，例如：

1. 调用一个分子生成模型，提出一批候选分子。
2. 调用ADMET（吸收、分布、代谢、排泄和毒性）预测模型，过滤掉性质不佳的分子。
3. 调用合成可行性评估工具，筛选出易于合成的分子。
4. 分析结果，如果都不满意，则调整生成策略，回到步骤1。

在这个过程中，智能体需要理解自然语言描述的目标，决定每一步该调用哪个工具（或模型），处理不同工具之间的输入输出格式转换，并根据中间结果动态调整后续计划。这更像一个虚拟的、不知疲倦的研发助手。

1.2 智能体系统的关键组件

一个典型的用于药物发现的AI智能体系统通常包含以下核心组件：

• 规划器：负责解析用户指令（如“寻找抗癌先导化合物”），并将其分解为一系列可执行的具体任务序列。这通常由一个大语言模型驱动。
• 工具集：智能体可以调用的外部能力集合。在药物发现中，工具可能包括：
  • 分子生成模型：如基于Transformer或扩散模型的分子生成器。
  • 性质预测模型：预测LogP、溶解度、毒性等。
  • 知识库查询：访问内部化合物数据库或公开的科研文献数据库。
  • 计算化学服务：调用分子对接模拟或量子化学计算。
• 执行器：负责实际调用工具，处理输入参数，并解析返回结果。
• 记忆模块：存储对话历史、任务上下文、以及从过往任务中学到的经验（例如，某种分子骨架总是导致高毒性），用于指导未来的决策。
• 评估与反馈循环：对工具执行的结果进行评估，判断是否满足目标，并决定下一步行动（继续、调整或终止）。

AWS与诺和诺德的合作，很可能是在AWS云上构建了一个集成这些组件的平台，利用AWS Bedrock提供的大语言模型作为智能体的“大脑”，利用Amazon SageMaker托管专业的药物发现模型作为“工具”，并通过AWS Step Functions或自定义编排逻辑来协调整个工作流。

2. 环境准备与AWS服务选型

构建一个原型系统，我们不需要从零开始训练所有模型。利用云服务提供的托管模型和计算资源是最高效的方式。以下是为模拟项目准备的环境和AWS服务。

2.1 AWS账户与基础服务开通

首先，你需要一个可用的AWS账户，并确保在目标区域（例如us-east-1或ap-northeast-1）拥有以下服务的访问权限：

1. IAM：创建具有特定权限的角色和策略，遵循最小权限原则。切勿使用根账户密钥进行编程访问。
2. Amazon Bedrock：这是AWS全托管的生成式AI服务，我们将使用它提供的Claude或Llama等大语言模型作为智能体的规划核心。你需要在AWS控制台为Bedrock“启用”你想要使用的模型。
3. Amazon SageMaker：用于部署和运行我们自定义的化学信息学模型（作为工具）。我们将使用SageMaker的实时推理端点。
4. AWS Lambda：用于编写轻量级的工具调用逻辑和格式转换代码。
5. Amazon API Gateway：为智能体系统提供统一的HTTP API接口。
6. Amazon S3：存储分子结构文件、模型数据、任务日志等。

2.2 项目依赖与本地开发环境

在本地开发机器上，我们需要配置Python环境来编写和测试智能体逻辑。

# 创建项目目录并初始化虚拟环境 mkdir drug-discovery-agent && cd drug-discovery-agent python3 -m venv .venv source .venv/bin/activate # Linux/macOS # .venv\Scripts\activate # Windows # 安装核心Python库 pip install boto3>=1.34.0 # AWS SDK pip install requests>=2.31.0 # HTTP请求 pip install pandas>=2.0.0 # 数据处理 pip install rdkit>=2023.9.5 # 化学信息学处理（核心） pip install pydantic>=2.0.0 # 数据验证 pip install langchain>=0.1.0 # （可选）用于快速构建智能体框架

注意：rdkit的安装在不同操作系统上可能比较复杂。在macOS上推荐使用conda install -c conda-forge rdkit。在Linux上可能需要从源码编译或寻找合适的wheel包。这是药物发现AI项目中最常见的基础依赖之一。

2.3 模拟工具：简单的性质预测模型

由于我们无法部署真实的药物发现模型，我们将创建一个极其简化的模拟模型。这个模型接收一个分子SMILES字符串，随机“预测”其类药性分数和合成难度分数。在实际项目中，这里应该替换为训练好的、部署在SageMaker上的真实模型。

创建一个文件simulated_models.py：

import random from typing import Dict from pydantic import BaseModel class MoleculeProperties(BaseModel): """分子性质模拟数据模型""" smiles: str drug_likeness_score: float # 类药性分数，0-1 synthetic_accessibility_score: float # 合成可及性分数，1-10（越低越易合成） predicted_activity: str # 预测活性，如 “High”, “Medium”, “Low” class PropertyPredictor: """模拟性质预测器""" def predict(self, smiles: str) -> MoleculeProperties: """ 模拟预测过程。 实际项目中，这里会调用SageMaker端点或本地模型。 """ # 简单的基于SMILES长度的随机“预测”，仅用于演示 seed = sum(ord(c) for c in smiles) random.seed(seed) drug_likeness = random.uniform(0.3, 0.95) syn_access = random.uniform(1.0, 8.0) # 根据分数简单判断活性 if drug_likeness > 0.7 and syn_access < 4.0: activity = "High" elif drug_likeness > 0.5: activity = "Medium" else: activity = "Low" return MoleculeProperties( smiles=smiles, drug_likeness_score=round(drug_likeness, 2), synthetic_accessibility_score=round(syn_access, 2), predicted_activity=activity ) # 示例用法 if __name__ == "__main__": predictor = PropertyPredictor() result = predictor.predict("CCO") print(result.model_dump_json(indent=2))

这个模拟器将在后续步骤中作为我们的一个“工具”被智能体调用。

3. 构建核心智能体工作流

我们将构建一个简化的工作流：用户提出需求，智能体规划任务，调用工具分析一批预定义的候选分子，并返回筛选结果。

3.1 定义智能体工具与调用接口

智能体需要知道它能用什么工具。我们首先定义工具的接口。创建agent_tools.py：

import json from abc import ABC, abstractmethod from typing import List, Any from simulated_models import PropertyPredictor, MoleculeProperties class BaseTool(ABC): """工具基类""" name: str description: str @abstractmethod def execute(self, input_data: Any) -> str: """执行工具，返回结果字符串（供LLM理解）""" pass class MoleculePropertyTool(BaseTool): """分子性质预测工具""" def __init__(self): self.name = "molecule_property_predictor" self.description = "Predicts drug-likeness, synthetic accessibility, and activity for a given molecule SMILES string." self.predictor = PropertyPredictor() def execute(self, input_data: str) -> str: # 输入可以是单个SMILES或列表 try: data = json.loads(input_data) if isinstance(data, list): smiles_list = data else: smiles_list = [str(data)] except json.JSONDecodeError: # 如果不是JSON，当作单个SMILES处理 smiles_list = [input_data] results = [] for smiles in smiles_list: props = self.predictor.predict(smiles) results.append(props.model_dump()) # 将结果格式化为LLM易于理解的文本 return json.dumps(results, indent=2) class MoleculeFilterTool(BaseTool): """分子筛选工具（基于规则）""" def __init__(self): self.name = "molecule_filter" self.description = "Filters a list of molecules based on property thresholds. Input must be a JSON list of molecule property objects." def execute(self, input_data: str) -> str: try: molecules = json.loads(input_data) except: return "Error: Input must be a valid JSON list of molecule properties." filtered = [] for mol in molecules: # 简单的筛选规则：类药性>0.6，合成难度<5.0 if mol.get('drug_likeness_score', 0) > 0.6 and mol.get('synthetic_accessibility_score', 10) < 5.0: filtered.append(mol) return json.dumps({ "original_count": len(molecules), "filtered_count": len(filtered), "filtered_molecules": filtered }, indent=2) # 工具注册表 TOOL_REGISTRY = { "property_predictor": MoleculePropertyTool(), "molecule_filter": MoleculeFilterTool(), } def get_tool(tool_name: str) -> BaseTool: return TOOL_REGISTRY.get(tool_name)

3.2 实现基于AWS Bedrock的规划器

智能体的“大脑”是一个大语言模型。我们将使用AWS Bedrock的Claude 3 Sonnet模型来解析用户指令并生成工具调用计划。创建bedrock_planner.py：

import boto3 import json from typing import List, Dict, Any class BedrockPlanner: def __init__(self, region_name='us-east-1', model_id='anthropic.claude-3-sonnet-20240229-v1:0'): self.bedrock_runtime = boto3.client('bedrock-runtime', region_name=region_name) self.model_id = model_id def create_plan(self, user_query: str, available_tools: List[Dict]) -> Dict[str, Any]: """ 根据用户查询和可用工具，生成执行计划。 返回一个包含工具调用序列的JSON结构。 """ tools_description = "\n".join([f"- {t['name']}: {t['description']}" for t in available_tools]) prompt = f"""你是一个药物发现AI助手。请根据用户需求，规划一个使用以下工具的执行序列。 可用的工具： {tools_description} 用户需求：{user_query} 请输出一个JSON数组，每个元素代表一个步骤，包含以下字段： - `step`: 步骤序号（从1开始） - `tool`: 要使用的工具名称（必须从上述工具列表中选择） - `input`: 该步骤的输入数据（可以是字符串或JSON）。如果输入依赖于上一步的输出，请使用 `$previous_output` 表示。 - `reason`: 选择此工具和输入的原因 示例： [ {{ "step": 1, "tool": "property_predictor", "input": ["CC(=O)OC1=CC=CC=C1C(=O)O", "CN1C=NC2=C1C(=O)N(C(=O)N2C)C"], "reason": "用户提供了候选分子SMILES，第一步需要预测它们的性质。" }}, {{ "step": 2, "tool": "molecule_filter", "input": "$previous_output", "reason": "根据预测的性质，筛选出类药性好且易于合成的分子。" }} ] 现在，请为以下需求生成计划： """ body = json.dumps({ "anthropic_version": "bedrock-2023-05-31", "max_tokens": 1000, "messages": [{ "role": "user", "content": prompt }] }) try: response = self.bedrock_runtime.invoke_model( modelId=self.model_id, body=body ) response_body = json.loads(response['body'].read()) plan_text = response_body['content'][0]['text'] # 提取JSON部分（模型可能返回带Markdown的文本） import re json_match = re.search(r'\[.*\]', plan_text, re.DOTALL) if json_match: plan = json.loads(json_match.group()) return {"plan": plan, "raw_response": plan_text} else: # 如果解析失败，返回原始文本 return {"plan": [], "raw_response": plan_text, "error": "Failed to parse JSON plan"} except Exception as e: return {"plan": [], "error": str(e)}

3.3 编排执行引擎

规划器生成计划后，需要一个执行引擎来按顺序调用工具，并处理步骤间的数据传递。创建execution_engine.py：

import json from typing import Dict, Any from agent_tools import get_tool class ExecutionEngine: def __init__(self): self.execution_history = [] def execute_plan(self, plan: List[Dict], initial_context: Dict = None) -> Dict[str, Any]: """ 执行给定的计划。 """ context = initial_context or {} final_output = None for step in plan: step_num = step.get('step') tool_name = step.get('tool') raw_input = step.get('input') reason = step.get('reason', '') print(f"[Step {step_num}] Using tool '{tool_name}': {reason}") # 解析输入：处理 $previous_output 变量 tool_input = self._resolve_input(raw_input, final_output, context) # 获取并执行工具 tool = get_tool(tool_name) if not tool: error_msg = f"Tool '{tool_name}' not found." self.execution_history.append({ "step": step_num, "status": "error", "error": error_msg }) return {"status": "error", "step": step_num, "error": error_msg} try: # 工具输入需要是字符串 if not isinstance(tool_input, str): tool_input_str = json.dumps(tool_input) else: tool_input_str = tool_input step_output = tool.execute(tool_input_str) final_output = step_output self.execution_history.append({ "step": step_num, "tool": tool_name, "input": tool_input, "output": step_output, "status": "success" }) print(f" Input: {tool_input_str[:100]}...") print(f" Output length: {len(step_output)} chars") except Exception as e: error_msg = f"Tool execution failed: {str(e)}" self.execution_history.append({ "step": step_num, "status": "error", "error": error_msg }) return {"status": "error", "step": step_num, "error": error_msg} return { "status": "success", "final_output": final_output, "history": self.execution_history } def _resolve_input(self, raw_input, previous_output, context): """解析输入，替换变量引用""" if isinstance(raw_input, str) and raw_input.strip() == "$previous_output": return previous_output # 可以扩展更多变量，如 $context.some_key return raw_input

4. 整合与运行：端到端智能体流程

现在我们将所有组件整合起来，创建一个可以接受用户查询、规划、执行并返回结果的完整流程。创建main.py作为入口点：

import json from bedrock_planner import BedrockPlanner from execution_engine import ExecutionEngine from agent_tools import TOOL_REGISTRY def main(): # 1. 初始化组件 print("Initializing Drug Discovery Agent...") planner = BedrockPlanner() engine = ExecutionEngine() # 2. 定义可用工具（供规划器知晓） available_tools = [ {"name": "property_predictor", "description": "Predicts molecular properties from SMILES."}, {"name": "molecule_filter", "description": "Filters molecules based on drug-likeness and synthetic accessibility scores."}, ] # 3. 模拟用户输入 # 在实际系统中，这可能来自API Gateway的HTTP请求 user_query = """ 我有一批候选分子，它们的SMILES如下： - CC(=O)OC1=CC=CC=C1C(=O)O (阿司匹林) - CN1C=NC2=C1C(=O)N(C(=O)N2C)C (咖啡因) - C1=CC=C(C=C1)C=O (苯甲醛) - CC(N)C (异丙胺) 请帮我分析这些分子的性质，并筛选出最有希望成为口服药物的分子。 """ print(f"User Query: {user_query[:200]}...\n") # 4. 调用Bedrock生成执行计划 print("Generating execution plan with Bedrock...") plan_result = planner.create_plan(user_query, available_tools) if "error" in plan_result: print(f"Planning failed: {plan_result['error']}") return plan = plan_result.get("plan", []) if not plan: print("No plan generated. Raw model response:") print(plan_result.get("raw_response", "Empty")) return print(f"Generated plan with {len(plan)} steps:") for step in plan: print(f" Step {step['step']}: {step['tool']} -> {step['reason']}") print() # 5. 执行计划 print("Executing plan...") # 我们可以从用户查询中提取初始数据（这里简单模拟） # 更智能的智能体会在规划时提取这些信息 initial_smiles = ["CC(=O)OC1=CC=CC=C1C(=O)O", "CN1C=NC2=C1C(=O)N(C(=O)N2C)C", "C1=CC=C(C=C1)C=O", "CC(N)C"] initial_context = {"candidate_smiles": initial_smiles} # 对于第一步，将SMILES列表作为输入 if plan and plan[0]['tool'] == 'property_predictor': plan[0]['input'] = initial_smiles execution_result = engine.execute_plan(plan, initial_context) # 6. 输出结果 print("\n" + "="*50) print("EXECUTION RESULT") print("="*50) if execution_result["status"] == "success": final_output = execution_result["final_output"] try: parsed_output = json.loads(final_output) print(json.dumps(parsed_output, indent=2)) except: print(final_output[:500]) # 打印前500字符 print(f"\nTotal steps executed: {len(execution_result['history'])}") else: print(f"Execution failed at step {execution_result.get('step')}:") print(execution_result.get("error", "Unknown error")) if __name__ == "__main__": main()

4.1 运行与验证

在配置好AWS凭证（通过aws configure或环境变量）并确保Bedrock模型已启用后，运行程序：

cd /path/to/drug-discovery-agent source .venv/bin/activate export AWS_DEFAULT_REGION=us-east-1 # 设置区域 python main.py

预期你会看到类似以下的输出（具体内容因模型响应而异）：

Initializing Drug Discovery Agent... User Query: 我有一批候选分子，它们的SMILES如下：... Generating execution plan with Bedrock... Generated plan with 2 steps: Step 1: property_predictor -> 用户提供了候选分子SMILES，第一步需要预测它们的性质。 Step 2: molecule_filter -> 根据预测的性质，筛选出类药性好且易于合成的分子。 Executing plan... [Step 1] Using tool 'property_predictor': 用户提供了候选分子SMILES，第一步需要预测它们的性质。 Input: ["CC(=O)OC1=CC=CC=C1C(=O)O", "CN1C=NC2=C1C(=O)N(C(=O)N2C)C", "C1=CC=C(C=C1)C=O", "CC(N)C"]... Output length: 892 chars [Step 2] Using tool 'molecule_filter': 根据预测的性质，筛选出类药性好且易于合成的分子。 Input: [{"smiles": "CC(=O)OC1=CC=CC=C1C(=O)O", "drug_likeness_score": 0.87, ...}]... Output length: 452 chars ================================================== EXECUTION RESULT ================================================== { "original_count": 4, "filtered_count": 2, "filtered_molecules": [ { "smiles": "CC(=O)OC1=CC=CC=C1C(=O)O", "drug_likeness_score": 0.87, "synthetic_accessibility_score": 3.21, "predicted_activity": "High" }, { "smiles": "CN1C=NC2=C1C(=O)N(C(=O)N2C)C", "drug_likeness_score": 0.72, "synthetic_accessibility_score": 4.85, "predicted_activity": "Medium" } ] } Total steps executed: 2

这个输出表明，智能体成功规划并执行了一个两步骤的工作流：先预测性质，再根据规则筛选，最终输出了两个符合“有希望成为口服药物”条件的分子。

5. 部署到AWS与生产化考量

原型在本地运行后，下一步是将其部署到AWS云上，使其成为一个可扩展、可靠的服务。

5.1 架构设计与服务部署

一个生产就绪的架构可能如下所示：

用户请求 -> Amazon API Gateway -> AWS Lambda (智能体协调器) | v AWS Step Functions (工作流编排) | v +----------------------+-----------------------+ | | | v v v Amazon Bedrock Amazon SageMaker Amazon DynamoDB (规划与决策) (专业模型推理端点) (存储任务状态与历史)

关键部署步骤：

1. 打包Lambda函数：将智能体协调逻辑（规划器、执行引擎）打包成Lambda部署包或容器镜像。
2. 部署SageMaker端点：将真实的性质预测模型（如基于DeepChem或GNN的模型）部署为SageMaker实时端点。
3. 创建Step Functions状态机：定义工作流，处理错误重试、超时和并行步骤。
4. 设置API Gateway：创建REST API，将HTTP请求路由到Lambda函数。
5. 配置IAM角色：确保每个服务都有最小必要权限访问其他服务（如Lambda调用Bedrock和Step Functions）。

5.2 关键配置与参数说明

在生产部署时，以下配置至关重要：

5.3 生产环境最佳实践与常见问题排查

将智能体系统投入实际药物研发，远不止让代码跑起来那么简单。

最佳实践：

1. 可观测性：在所有关键步骤添加结构化日志（使用Amazon CloudWatch Logs）。记录用户查询、生成的计划、每个工具调用的输入输出、执行时间以及最终结果。这对于调试和审计至关重要。
2. 数据安全与合规：药物研发数据高度敏感。
   • 加密：确保S3桶、SageMaker端点、Bedrock数据均使用KMS密钥加密。
   • 网络隔离：将SageMaker端点部署在私有VPC中，仅允许Lambda通过VPC端点访问。
   • 数据残留：明确Bedrock等服务的日志和数据保留策略，确保符合公司政策。
3. 成本控制：
   • 缓存：对常见的分子性质预测结果进行缓存（如使用Amazon ElastiCache），避免重复计算。
   • 监控预算：为Bedrock和SageMaker设置CloudWatch预算告警。
   • 选择模型：在原型阶段使用成本更低的模型（如Haiku），在关键验证阶段再换用更强模型。
4. 错误处理与重试：
   • 网络超时、模型服务限流、临时性错误是常态。在Lambda和Step Functions中实现指数退避重试机制。
   • 为智能体规划设定“最大步数”限制，防止因逻辑错误导致无限循环。

常见问题排查清单：

6. 扩展方向与未来展望

本文展示的只是一个高度简化的原型。真实的药物发现智能体系统要复杂得多。以下是几个关键的扩展方向：

1. 集成真实模型与工具：将模拟工具替换为真实的计算化学工具。例如，使用SageMaker部署：
   • 分子生成模型：如REINVENT、GENTRL。
   • ADMET预测模型：使用DeepChem或商业软件训练的模型。
   • 分子对接模拟：集成AutoDock Vina或Schrodinger的对接工具。
2. 引入反馈与强化学习：让智能体不仅执行计划，还能从历史任务的成功/失败中学习。例如，如果某种分子骨架频繁被过滤掉，智能体可以在后续的分子生成中主动避免它。这需要将执行结果与元数据存储到DynamoDB，并用于微调规划策略。
3. 多模态能力：未来的智能体可能需要处理不仅仅是SMILES字符串。它可以分析实验报告（PDF）、化学结构图（图像）、甚至科学文献（文本），形成更全面的决策依据。这需要集成多模态大模型。
4. 人机协同：设计智能体与化学家交互的界面。智能体可以提出建议，化学家可以给出反馈（“这个分子合成路线太长了”），智能体据此调整后续策略。这需要构建一个持续的学习循环。

构建药物发现AI智能体是一个系统工程，它考验的不仅是机器学习算法能力，更是对研发流程的深刻理解、复杂系统的架构设计以及云原生技术的娴熟运用。从本文的原型出发，深入每一个扩展方向，都将打开一扇通往更高效、更智能的药物研发未来之门。