AIGC 领域多智能体系统在科研领域的实验辅助应用

    import numpy as np
    from collections import defaultdict
    
    class MultiAgentSystem:
    def __init__(self, num_agents, state_space, action_space):
        self.num_agents = num_agents
        self.state_space = state_space
        self.action_space = action_space
        self.q_tables = [defaultdict(lambda: np.zeros(action_space))
                        for _ in range(num_agents)]
        self.alpha = 0.1  # 学习率
        self.gamma = 0.9  # 折扣因子
        self.epsilon = 0.1  # 探索率
    
    def choose_action(self, agent_idx, state):
        if np.random.random() < self.epsilon:
            return np.random.choice(self.action_space)
        return np.argmax(self.q_tables[agent_idx][state])
    
    def learn(self, agent_idx, state, action, reward, next_state):
        best_next_action = np.argmax(self.q_tables[agent_idx][next_state])
        td_target = reward + self.gamma * self.q_tables[agent_idx][next_state][best_next_action]
        td_error = td_target - self.q_tables[agent_idx][state][action]
        self.q_tables[agent_idx][state][action] += self.alpha * td_error
    
    def coordinate(self, states, actions, rewards, next_states):
        for i in range(self.num_agents):
            self.learn(i, states[i], actions[i], rewards[i], next_states[i])
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/Q1KUoSY5ZFyTBRr7Cx4MphH23fvA.png)

    from scipy.stats import norm
    import numpy as np
    
    class BayesianOptimizer:
    def __init__(self, bounds, init_points=5):
        self.bounds = bounds
        self.X = []
        self.y = []
        self.init_points = init_points
        self.dim = len(bounds)
    
    def acquisition_function(self, x, gp, xi=0.01):
        mu, sigma = gp.predict(np.array([x]), return_std=True)
        mu_sample = gp.predict(self.X)
        mu_sample_opt = np.max(mu_sample)
        with np.errstate(divide='warn'):
            imp = mu - mu_sample_opt - xi
            Z = imp / sigma
            ei = imp * norm.cdf(Z) + sigma * norm.pdf(Z)
            ei[sigma == 0.0] = 0.0
        return ei
    
    def optimize(self, objective_func, n_iter=10):
        from sklearn.gaussian_process import GaussianProcessRegressor
        from sklearn.gaussian_process.kernels import RBF
    
        kernel = RBF(length_scale=1.0)
        gp = GaussianProcessRegressor(kernel=kernel)
    
        # 初始随机采样
        for _ in range(self.init_points):
            x = np.array([np.random.uniform(b[0], b[1]) for b in self.bounds])
            y = objective_func(x)
            self.X.append(x)
            self.y.append(y)
    
        # 贝叶斯优化
        for _ in range(n_iter):
            gp.fit(self.X, self.y)
            x_next = self._next_point(gp)
            y_next = objective_func(x_next)
            self.X.append(x_next)
            self.y.append(y_next)
    
        best_idx = np.argmax(self.y)
        return self.X[best_idx], self.y[best_idx]
    
    def _next_point(self, gp):
        best_x = None
        best_acq = -np.inf
    
        # 在边界内随机采样候选点
        for _ in range(100):
            x = np.array([np.random.uniform(b[0], b[1]) for b in self.bounds])
            acq = self.acquisition_function(x, gp)
            if acq > best_acq:
                best_acq = acq
                best_x = x
        return best_x
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/xh3XOTaNwp7rt4PF5cRfZkeCsVuv.png)

    conda create -n mas-science python=3.8
    conda activate mas-science
    pip install torch gym rdkit scikit-learn pandas numpy
    
    
    bash

    import numpy as np
    from rdkit import Chem
    from rdkit.Chem import AllChem
    from collections import defaultdict
    
    class ChemistryExperimentAgent:
    def __init__(self, target_molecule):
        self.target = target_molecule
        self.q_table = defaultdict(lambda: np.zeros(100))  # 假设有100种可能的反应
    
    def evaluate_reaction(self, reaction_smile):
        # 使用RDKit评估反应效果
        reactant = Chem.MolFromSmiles(reaction_smile)
        if reactant is None:
            return -1.0
        fp1 = AllChem.GetMorganFingerprintAsBitVect(self.target, 2)
        fp2 = AllChem.GetMorganFingerprintAsBitVect(reactant, 2)
        similarity = AllChem.DataStructs.TanimotoSimilarity(fp1, fp2)
        return similarity
    
    def choose_reaction(self, state):
        # ε-贪婪策略选择反应
        if np.random.random() < 0.1:
            return np.random.randint(100)
        return np.argmax(self.q_table[state])
    
    def update_knowledge(self, state, action, reward, next_state):
        # Q-learning更新
        best_next_action = np.argmax(self.q_table[next_state])
        td_target = reward + 0.9 * self.q_table[next_state][best_next_action]
        td_error = td_target - self.q_table[state][action]
        self.q_table[state][action] += 0.1 * td_error
    
    class ChemistryMAS:
    def __init__(self, target_smile, num_agents=3):
        self.target = Chem.MolFromSmiles(target_smile)
        self.agents = [ChemistryExperimentAgent(self.target)
                      for _ in range(num_agents)]
        self.shared_knowledge = defaultdict(float)
    
    def run_experiment(self, num_episodes=100):
        for episode in range(num_episodes):
            # 每个智能体独立探索
            for agent in self.agents:
                state = np.random.randint(10)  # 初始状态
                action = agent.choose_reaction(state)
                reward = agent.evaluate_reaction(self._action_to_smile(action))
                next_state = np.random.randint(10)  # 简化状态转移
                agent.update_knowledge(state, action, reward, next_state)
    
                # 更新共享知识
                self.shared_knowledge[(state, action)] = max(
                    self.shared_knowledge[(state, action)], reward)
    
            # 知识共享阶段
            if episode % 10 == 0:
                self._share_knowledge()
    
    def _action_to_smile(self, action):
        # 简化版：将动作映射到SMILES字符串
        # 实际应用中需要更复杂的映射
        return f"CCO{action}"
    
    def _share_knowledge(self):
        # 智能体间共享最佳知识
        best_knowledge = max(self.shared_knowledge.values())
        for agent in self.agents:
            for (state, action), value in self.shared_knowledge.items():
                if value == best_knowledge:
                    agent.q_table[state][action] = max(
                        agent.q_table[state][action], value)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/GO3d07AvMfHPwyDguVbrX4iB95US.png)

 * 每个智能体维护自己的Q表(q_table)记录反应知识
 * 使用RDKit计算分子相似度作为奖励信号
 * 采用Q-learning算法更新知识

 * 管理多个化学实验智能体
 * 维护共享知识库(shared_knowledge)
 * 定期进行知识共享(_share_knowledge方法)

 * 每个智能体独立探索化学反应空间
 * 定期将最佳知识共享给所有智能体
 * 通过多轮迭代优化实验方案

 * 结合了强化学习和化学信息学
 * 实现了分布式探索和集中式知识共享
 * 可扩展性强，可添加更多专业智能体