FastCAE AI应用助手重构(一)架构与桥接详解

FastCAE 是一个 C++/Qt 的 CAE 平台，拥有成熟的几何、网格、求解器、后处理等能力；而当下的大模型生态几乎全部生长在 Python 里（LangChain、deepagents、MCP……）。

要让 AI 真正"用"起来，只把模型接进来是不够的，模型必须能操作平台：读数据、改模型、跑求解。于是有了一个很自然的架构取舍：

▍C++ 侧：负责 UI、组件接入、以及"真正执行命令"的平台能力；

▍Python 侧：负责大模型推理、Agent 编排、工具调用、技能加载。

两者之间用一条进程间协议连接，这就是本文的主角：stdio  JSONL桥接。

▲ 双引擎分工：平台能力 × 智能生态

▲ FITKDeepAgent 整体架构全景

FITKDeepAgentInterface.cpp

QString FITKDeepAgentInterface::getComponentName(){    return "FITKDeepAgent";           // 组件唯一标识，工厂据此查找}QWidget* FITKDeepAgentInterface::getWidget(const int indexPort){    Q_UNUSED(indexPort);    return new FITKDeepAgentWindow(); // 返回组件主窗口}

协议语义速览

请求（宿主 → agent）:    ping / chat / resume / reset / list_sessions / shutdown事件（agent → 宿主）:    pong / start / thinking / tool_call / tool_result /    answer / done / interrupt / error反向请求（agent → 宿主）:    hostpy —— agent 请求平台执行命令，    宿主回写 hostpy_result（按 id 关联）

startAgent() · 拉起子进程

void FITKDeepAgentClient::startAgent(){    QString pythonExe, agentScript;    if (!this->locateRuntime(pythonExe, agentScript))    {        emit sigFailed(QString("FastCAEDeepAgent runtime not found: %1").arg(_agentRoot));        return;    }    _process->setProgram(pythonExe);          // .venv/Scripts/python.exe    _process->setArguments({ agentScript });   // Agent.py    _process->setWorkingDirectory(_agentRoot); // FastCAEDeepAgent/    // 强制 UTF-8，避免 Windows 控制台默认编码破坏 JSONL 协议    QProcessEnvironment env = QProcessEnvironment::systemEnvironment();    env.insert("PYTHONUTF8", "1");    _process->setProcessEnvironment(env);    _process->start();}

slotReadStdout() · 逐行解析

void FITKDeepAgentClient::slotReadStdout(){    _process->setReadChannel(QProcess::StandardOutput);    while(_process->canReadLine())    {        const QByteArray raw = _process->readLine();        const QString line = QString::fromUtf8(raw).trimmed();        if(line.isEmpty()) continue;        QJsonParseError parseError;        QJsonDocument doc = QJsonDocument::fromJson(line.toUtf8(), &parseError);        if(parseError.error != QJsonParseError::NoError || !doc.isObject()) continue;        this->dispatch(doc.object());   // 合法事件交给分发器    }}

dispatch() · 事件分发

void FITKDeepAgentClient::dispatch(constQJsonObject& event){    const QString type = event.value("type").toString();    if (type == "hostpy")          this->handleHostPyRequest(event);    else if (type == "pong")       emit sigConnected();    else if (type == "thinking")   emit sigThinking(event.value("content").toString());    else if (type == "tool_call")  emit sigToolCall(event.value("name").toString(),                                                    event.value("args").toObject());    else if (type == "tool_result") emit sigToolResult(event.value("name").toString(),                                                       event.value("content").toVariant().toString());    else if (type == "answer")     emit sigAnswer(event.value("content").toString());    else if (type == "done")       emit sigDone();    else if (type == "interrupt")  emit sigInterrupt(event.value("session_id").toString(),                                                     event.value("request").toObject());    else if (type == "error")      emit sigFailed(event.value("error").toString());}

Agent.py · stdio 服务器入口

async def main() -> None:    sink = StdioSink()                  # 先抓原始 stdout 的二进制 buffer    sys.stdout = sys.stderr             # 之后所有 print 诊断都去 stderr    agent = FastCAEAgent(        config_ini=_ROOT / ".env" / "Agent.ini",        llm_ini=_ROOT / ".env" / "LLM.ini",        sink=sink,        extra_tools=[build_host_tool(sink)],   # 宿主桥接工具：平台内执行命令    )    await agent.ready()    await StdioTransport(agent).run()

FastCAEAgent.ready() · 组装流水线

async def ready(self) -> "FastCAEAgent":    llm = LLMProxy(self.llm_ini).build(self.config.llm_section)   # 1. 模型    tools = await MCPProxy(self.config.mcp_dir).load_tools()      # 2. MCP 工具    if self.extra_tools:        tools = tools + self.extra_tools                          # 3. 宿主桥接工具    skills = SkillsProxy(self.config.skills_dir, self.config.backend_root)    interrupt_on = {name: True for name in self.config.hitl_tools} # 4. HITL 白名单    middleware = [EventMiddleware(self.sink)] if self.sink else [] # 5. 事件中间件    self.agent = create_deep_agent(        model=llm,        system_prompt=self.config.system_prompt,        tools=tools,        middleware=middleware,        checkpointer=self.checkpointer,   # InMemorySaver：会话历史存进程内存        backend=skills.backend,        skills=skills.source_paths(),        interrupt_on=interrupt_on or None,    )    return self

StdioSink · 事件写出与反向请求同步

class StdioSink:    def emit(self, event: dict) -> None:        with self._lock:            event = {"timestamp": _now_iso(), **event}            line = json.dumps(event, ensure_ascii=False) + "\n"            self._out.write(line.encode("utf-8"))            self._out.flush()    def request(self, payload: dict, timeout=300.0) -> dict:        """向宿主发起一次反向请求并阻塞等待应答（供工具线程调用）。"""        with self._lock:            self._reqSeq += 1            req_id = self._reqSeq            line = json.dumps({"id": req_id, **payload}, ensure_ascii=False) + "\n"            self._out.write(line.encode("utf-8"))            self._out.flush()            done = threading.Event()            holder = [None]            self._pending[req_id] = (done, holder)        if not done.wait(timeout):            return {"ok": False, "error": f"host request timeout after {timeout}s"}        return holder[0]    def resolve(self, response: dict) -> bool:        """读线程收到宿主应答时调用：按 id 唤醒等待者。"""        req_id = response.get("id")        with self._lock:            slot = self._pending.pop(req_id, None)        if slot is None:            return False        slot[1][0] = response        slot[0].set()        return True

▍emit：可能来自事件循环（chat 事件流）和工具线程（hostpy 反向请求）两个线程，内部用互斥锁保护；

▍反向请求用 id 关联 + threading.Event 做同步：发出请求后阻塞等待，收到 hostpy_result 时按 id 唤醒对应的等待线程。

StdioTransport._dispatch() · 请求循环

def _dispatch(self, loop, sink, req) -> bool:    type_ = req.get("type")    if type_ == "hostpy_result":        sink.resolve(req)                    # 反向应答：唤醒等待的工具线程    elif type_ == "ping":        sink.emit({"type": "pong", "ok": True})    elif type_ == "chat":        sid = req.get("session_id") or "默认"        future = asyncio.run_coroutine_threadsafe(            self._run_turn(sid, req.get("message") or "", chat=True), loop)        future.add_done_callback(self._swallow_turn_error)    elif type_ == "resume":        sid = req.get("session_id") or "默认"        future = asyncio.run_coroutine_threadsafe(            self._run_turn(sid, req.get("decisions") or [], chat=False), loop)        future.add_done_callback(self._swallow_turn_error)    elif type_ == "reset":        self.agent.sessions.delete(req.get("session_id"))        sink.emit({"type": "reset", "ok": True, "session_id": req.get("session_id")})    elif type_ == "list_sessions":        sink.emit({"type": "list_sessions", "sessions": self.agent.sessions.list_ids()})    elif type_ == "shutdown":        sink.emit({"type": "shutdown", "ok": True})        return True    else:        sink.emit({"type": "error", "ok": False, "error": f"未知请求类型: {type_}"})    return False

▲ hostpy 反向请求完整链路

hostbridge.py · 宿主桥接工具

def build_host_tool(sink, timeout=300.0):    @tool    def submit_python_command(command: str) -> str:        """Submit one Python command line to the FastCAE host platform for execution.        Use this tool to run a single Python statement inside the FastCAE        application's embedded interpreter, e.g. to drive the platform's        data model or API. Returns the evaluated result text on success, or        an error message on failure. Submit one statement per call and call        the tool repeatedly for multi-step scripts.        """        response = sink.request({"type": "hostpy", "command": command}, timeout=timeout)        if response.get("ok"):            return response.get("result") or "OK"        return f"ERROR: {response.get('error')}"    return submit_python_command

handleHostPyRequest() · 接收反向请求

voidFITKDeepAgentClient::handleHostPyRequest(const QJsonObject& request){    const QString command = request.value("command").toString();    const int requestId   = request.value("id").toInt();    // 命令经共用执行器提交宿主内嵌解释器    QString errorInfo, result;    const bool ok = FITKDeepAgentCommandExecutor::executeCommandFromText(        command, errorInfo, &result);    // 按 id 关联回写应答：成功携带 result，失败携带 error    QJsonObject response;    response.insert("type", "hostpy_result");    response.insert("id", requestId);    response.insert("ok", ok);    if (ok) response.insert("result", result);    else    response.insert("error", errorInfo);    this->sendJson(response);    emit sigHostCommand(command, ok, ok ? result : errorInfo);}

executeCommandFromText() · 命令执行器

boolFITKDeepAgentCommandExecutor::executeCommandFromText(    const QString& commandText, QString& errorInfo, QString* result){    QString statement = commandText.trimmed();    if (statement.isEmpty()) return true;   // 空命令视为成功    // 兼容赋值语句："a = b" → "(a := b)"，让赋值也能作为表达式求值返回结果    staticconst QRegularExpression assignmentPattern(        "^([A-Za-z_][A-Za-z0-9_]*)\\s*=\\s*(.+)$");    const QRegularExpressionMatch m = assignmentPattern.match(statement);    if (m.hasMatch())        statement = QString("(%1 := %2)").arg(m.captured(1), m.captured(2).trimmed());    else if (statement.contains(":=")             && !(statement.startsWith('(') && statement.endsWith(')')))        statement = QString("(%1)").arg(statement);    // 提交宿主内嵌解释器（FITKPython，基于 PythonQt）    Python::FITKPythonInterface* py = Python::FITKPythonInterface::getInstance();    if (py == nullptr) { errorInfo = QObject::tr("FITKPythonInterface is null."); return false; }    QVariant pyResult;    if (!py->submit(statement, pyResult))    {        errorInfo = py->getErrorInfo();      // 优先返回解释器给出的错误        return false;    }    if (result != nullptr && pyResult.isValid() && !pyResult.isNull())        *result = pyResult.toString();    return true;}

▲ HITL 人工审批流程

Agent.ini · HITL 白名单

[hitl]; 逗号分隔，这些工具执行前需人工审批；留空则关闭 HITLtools = submit_python_command

slotOnInterrupt() · 审批入口

void FITKDeepAgentWindow::slotOnInterrupt(const QString& sessionId, const QJsonObject& request){    // 弹出审批对话框收集决定，取消时全部拒绝，最后 resume 回传    FITKDeepAgentHitlDialog dlg(request, this);    const QJsonArray decisions = (dlg.exec() == QDialog::Accepted)        ? dlg.getDecisions()        : /* 取消 → 全部拒绝 */;    _client->resume(sessionId, decisions);}

▍进程边界即信任边界。大模型推理跑在独立 Python 子进程里，平台核心跑在 C++ 里，中间只隔一条可审计的 JSONL 协议流——AI 想动平台，只能走 hostpy 这条被审批（HITL）管住的路。

▍传输与引擎解耦。EventSink 是框架唯一的抽象点，Agent.py / stdio_transport.py 只是 stdio 这一种传输的应用层示例。将来换 http/websocket，框架一行不改。

▍一条协议流，双向请求。stdout 既承载 agent 的事件，也承载 agent 的反向请求（hostpy），用 id 关联 + threading.Event 优雅地解决了跨进程双向同步，同时用独立读线程避免了双向请求死锁。

▍两条入口行为一致。命令执行器 FITKDeepAgentCommandExecutor 被 stdio 桥接与 HTTP 桥接共用，平台能力只写一份。