s08

IPC 通信

仿真层

文件系统跨进程通信机制

基于 JSON 文件的命令/响应 IPC 模式，支持智能体实时访谈

选择文件系统而非 Socket 作为 IPC 通道，降低复杂度的同时支持跨平台兼容——简单即可靠

📡

Flask 主进程如何与 OASIS 仿真子进程实时通信？

仿真运行在独立子进程中，但用户可能随时想采访某个智能体或关闭仿真环境。传统的 Socket/HTTP IPC 对这个场景过于复杂。MiroFish 选择了最简单的方案——基于文件系统的命令/响应模式，用 JSON 文件在目录间传递消息。

Flask 主进程 (Client)

c1|send_command() 写入命令 JSON

c2|轮询 ipc_responses/ 目录

c3|读取并删除响应文件

c4|返回 IPCResponse 给 API 层

←ipc_commands/{id}.json→

←ipc_responses/{id}.json→

←命令执行完成→

←UUID 一一对应→

OASIS 子进程 (Server)

s1|poll_commands() 扫描命令目录

s2|反序列化 IPCCommand

s3|执行采访/关闭等操作

s4|send_success() 写入响应 JSON

映射关系

c1 → s1: ipc_commands/{id}.json

c2 → s4: ipc_responses/{id}.json

c3 → s3: 命令执行完成

c4 → s2: UUID 一一对应

总览

IPC 模块实现了 Flask 主进程和 OASIS 仿真子进程之间的双向通信。采用文件系统方案：Client 写命令 JSON → Server 轮询读取并执行 → Server 写响应 JSON → Client 轮询读取。支持三种命令：采访(INTERVIEW)、批量采访(BATCH_INTERVIEW)、关闭环境(CLOSE_ENV)。

1/6

Key Source Code

Client 端：写入命令 JSON 文件，轮询响应文件

backend/app/services/simulation_ipc.py

class SimulationIPCClient:
    def send_command(self, command_type, args, timeout=60.0, poll_interval=0.5):
        command_id = str(uuid.uuid4())
        command = IPCCommand(command_id=command_id, command_type=command_type, args=args)

        # 写入命令文件到 ipc_commands/ 目录
        command_file = os.path.join(self.commands_dir, f"{command_id}.json")
        with open(command_file, 'w', encoding='utf-8') as f:
            json.dump(command.to_dict(), f, ensure_ascii=False)

        # 轮询等待响应文件出现在 ipc_responses/ 目录
        response_file = os.path.join(self.responses_dir, f"{command_id}.json")
        while time.time() - start_time < timeout:
            if os.path.exists(response_file):
                response = IPCResponse.from_dict(json.load(open(response_file)))
                os.remove(command_file)   # 清理
                os.remove(response_file)
                return response
            time.sleep(poll_interval)
        raise TimeoutError(f"等待命令响应超时 ({timeout}秒)")

python

Server 端：轮询命令目录，执行后写入响应

backend/app/services/simulation_ipc.py

class SimulationIPCServer:
    def poll_commands(self) -> Optional[IPCCommand]:
        """轮询 commands/ 目录，按时间排序返回第一个待处理命令"""
        command_files = []
        for filename in os.listdir(self.commands_dir):
            if filename.endswith('.json'):
                filepath = os.path.join(self.commands_dir, filename)
                command_files.append((filepath, os.path.getmtime(filepath)))
        command_files.sort(key=lambda x: x[1])  # 按修改时间排序
        for filepath, _ in command_files:
            data = json.load(open(filepath))
            return IPCCommand.from_dict(data)
        return None

    def send_success(self, command_id, result):
        """写入成功响应，同时删除命令文件"""
        response_file = os.path.join(self.responses_dir, f"{command_id}.json")
        json.dump(IPCResponse(command_id, COMPLETED, result).to_dict(), ...)

python