Python 上位机 + Claude Code 实现试剂研发全自动迭代闭环系统

在试剂研发场景中，传统实验依赖人工记录数据、手动调整配方，效率低且易出错。本文将基于，搭建一套的智能实验闭环系统，实现：设备数据自动上传 → AI 分析迭代 → 自动下发配方指令 → 循环优化，关闭程序重启后仍可继承历史实验记忆。

ZHANG13HAO

389人浏览 · 2026-04-09 10:29:58

ZHANG13HAO · 2026-04-09 10:29:58 发布

在试剂研发场景中，传统实验依赖人工记录数据、手动调整配方，效率低且易出错。本文将基于 Python 上位机 + 串口通信 + Claude Code，搭建一套带永久记忆、自动迭代优化的智能实验闭环系统，实现：设备数据自动上传 → AI 分析迭代 → 自动下发配方指令 → 循环优化，关闭程序重启后仍可继承历史实验记忆。

一、系统整体流程

整个系统为全自动闭环逻辑，无需人工干预，流程如下：

实验设备通过串口将实时试剂数据（浓度、温度、反应结果等）上传至 Python 上位机；
上位机接收数据并格式化，发送给常驻运行的 Claude Code；
Claude Code 保留上下文记忆，基于历史实验数据迭代分析，给出试剂用量调整指令；
上位机接收 AI 调整结果，通过串口下发给实验设备执行；
每轮迭代数据自动保存至 history.md，关闭程序重启后可恢复记忆，继续迭代。

二、核心原理说明

1. Claude Code 上下文记忆机制

进程常驻：Claude Code 仅启动一次且不关闭，全程保留对话上下文，实现连续迭代优化；
永久记忆方案：Claude Code 进程关闭后记忆会丢失，因此通过 history.md 文件持久化所有实验记录，重启后自动加载历史，实现无缝续跑。

2. 串口通信逻辑

使用 Python pyserial 库实现上位机与实验设备的双向通信：
- 读取设备上传的实验数据；
- 发送 AI 生成的试剂调整指令至设备。

3. 自动迭代逻辑

上位机通过 subprocess 启动 Claude Code 子进程，通过标准输入输出（stdin/stdout）持续交互；
每轮实验数据 + 调整结果实时写入历史文件，保证数据不丢失。

三、环境准备

1. 安装依赖库

bash

运行

# 串口通信库
pip install pyserial

2. 环境要求

已安装并配置好 Claude Code CLI，命令行可直接调用 claude code；
实验设备与电脑正常串口连接，知晓串口号（如 COM3）、波特率（如 9600）；
Python 3.8 及以上版本。

四、完整实现代码

python

运行

import serial
import subprocess
from subprocess import PIPE
import time

# ====================== 系统配置项 ======================
# 串口配置
SERIAL_PORT = "COM3"       # 实验设备串口号
BAUD_RATE = 9600          # 设备波特率
SERIAL_TIMEOUT = 2        # 串口超时时间

# 历史记忆文件（关闭程序后永久保存）
HISTORY_FILE = "history.md"

# AI 交互配置
AI_SYSTEM_PROMPT = """
你是专业试剂研发智能优化助手，负责全程迭代优化试剂配方。
严格遵循以下规则：
1. 记住所有历史实验数据和调整记录，基于历史做精准迭代；
2. 仅返回试剂用量调整指令，格式示例：试剂A=5.2ml，试剂B=3.1ml；
3. 不输出多余文字、解释、标点外的符号，仅保留最简指令；
4. 结合最新实验数据，逐步优化试剂配比，提升实验效果。
"""
# ======================================================

# ====================== 历史记忆管理 ======================
def load_experiment_history() -> str:
    """加载历史实验记录，重启后恢复AI记忆"""
    try:
        with open(HISTORY_FILE, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        # 文件不存在则创建并返回空历史
        with open(HISTORY_FILE, "w", encoding="utf-8") as f:
            f.write("# 试剂研发实验历史记录\n")
        return "暂无历史实验数据"

def append_to_history(experiment_data: str, adjust_command: str) -> None:
    """追加每轮迭代数据到历史文件，实现永久记忆"""
    record = f"""
---
### 迭代时间：{time.strftime('%Y-%m-%d %H:%M:%S')}
实验数据：{experiment_data}
AI调整指令：{adjust_command}
"""
    with open(HISTORY_FILE, "a", encoding="utf-8") as f:
        f.write(record)

# ====================== Claude Code 启动 ======================
def start_claude_code() -> subprocess.Popen:
    """启动常驻 Claude Code 进程，保留上下文记忆"""
    claude_process = subprocess.Popen(
        ["claude", "code"],
        stdin=PIPE,
        stdout=PIPE,
        stderr=PIPE,
        text=True,
        encoding="utf-8",
        bufsize=1  # 行缓冲，保证实时交互
    )

    # 加载历史记忆 + 发送系统提示词
    history_content = load_experiment_history()
    init_prompt = f"{AI_SYSTEM_PROMPT}\n历史实验记录：\n{history_content}\n"

    claude_process.stdin.write(init_prompt)
    claude_process.stdin.flush()
    print("✅ Claude Code 启动完成，历史记忆已加载")
    return claude_process

# ====================== 串口通信 ======================
def init_serial_port() -> serial.Serial:
    """初始化串口，连接实验设备"""
    try:
        ser = serial.Serial(
            port=SERIAL_PORT,
            baudrate=BAUD_RATE,
            timeout=SERIAL_TIMEOUT
        )
        if ser.is_open:
            print(f"✅ 串口 {SERIAL_PORT} 连接成功")
        return ser
    except Exception as e:
        print(f"❌ 串口连接失败：{str(e)}")
        raise SystemExit(1)

def read_device_data(ser: serial.Serial) -> str:
    """读取设备上传的实验数据"""
    raw_data = ser.readline().decode("utf-8", errors="ignore").strip()
    return raw_data

def send_adjust_to_device(ser: serial.Serial, command: str) -> None:
    """向设备发送AI调整后的试剂用量指令"""
    send_data = command + "\n"
    ser.write(send_data.encode("utf-8"))

# ====================== 全自动迭代主循环 ======================
def auto_iteration_loop():
    """启动全自动试剂研发迭代闭环"""
    print("=" * 60)
    print("🚀 试剂研发全自动迭代系统启动")
    print("=" * 60)

    # 初始化模块
    claude = start_claude_code()
    ser = init_serial_port()

    try:
        while True:
            # 1. 读取设备实验数据
            exp_data = read_device_data(ser)
            if not exp_data:
                time.sleep(0.5)
                continue

            print(f"\n📥 最新实验数据：{exp_data}")

            # 2. 发送数据给 Claude Code 迭代分析
            claude.stdin.write(f"最新实验数据：{exp_data}\n")
            claude.stdin.flush()

            # 3. 获取AI调整指令
            adjust_cmd = claude.stdout.readline().strip()
            print(f"🤖 AI 迭代调整：{adjust_cmd}")

            # 4. 保存本轮数据到历史文件
            append_to_history(exp_data, adjust_cmd)

            # 5. 下发指令给实验设备
            send_adjust_to_device(ser, adjust_cmd)
            print("📤 已下发调整指令至实验设备")

            # 等待设备执行完成，进入下一轮迭代
            time.sleep(3)

    except KeyboardInterrupt:
        print("\n🛑 系统手动停止，所有实验数据已保存至 history.md")
    finally:
        ser.close()
        claude.terminate()
        print("🔌 串口与 Claude Code 已安全关闭")

# ====================== 启动系统 ======================
if __name__ == "__main__":
    auto_iteration_loop()