必须显式声明并发执行上下文与资源隔离策略，否则langgraph自动串行化导致吞吐量归零；需升级langgraph≥0.1.47、封装atom为runnablelambda、用runnableparallel构建并发图、设stream_mode="values"并传入含batch_inputs键的字典。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
要在elizaOS中实现Atoms多模型同时调用并批量执行任务，必须绕过单原子串行调度的默认行为，显式声明并发执行上下文与资源隔离策略，否则所有Atoms将被LangGraph自动串行化，导致吞吐量归零。
准备并发执行环境
进入elizaOS项目根目录，确认已激活支持异步I/O的Python环境（≥3.10），并检查LangGraph版本是否为≥0.1.47——旧版本不支持RunnableParallel在stateful graph中的嵌套调用。
运行pip install --upgrade langgraph==0.1.47，升级后验证：python -c "from langgraph.graph import RunnableParallel; print('OK')"。若报错AttributeError，则说明未升级成功，【必须重装】。
创建concurrent_atoms.py文件，导入核心模块：from langgraph.graph import StateGraph, START, END和from langgraph.prebuilt import ToolNode。
定义可并发调用的Atoms
每个Atom必须封装为独立的RunnableLambda，且内部不得共享全局状态或mutable对象。例如：
方法一：封装本地模型Atom
定义atom_local_llm = RunnableLambda(lambda x: {"response": local_inference(x["input"])})，其中local_inference函数需使用torch.inference_mode()并显式指定device="cuda:0"——若漏写设备参数，PyTorch会默认加载到CPU，造成GPU空转。
方法二：封装API模型Atom
使用httpx.AsyncClient而非requests，构造atom_api_llm = RunnableLambda(lambda x: asyncio.run(call_remote_api(x["input"])))。注意：不能直接在lambda里用await，必须包裹进asyncio.run()，否则LangGraph调度器会阻塞整个事件循环。
构建并发图结构
第一步：初始化状态图graph = StateGraph(dict)，该dict必须包含"batch_inputs"字段，用于接收批量输入列表。
第二步：注册并发节点graph.add_node("parallel_invoke", RunnableParallel({"model_a": atom_local_llm, "model_b": atom_api_llm}))。此处键名"model_a"和"model_b"将成为输出字典的顶层key，不可重复或含空格。
第三步：设置边路逻辑graph.add_edge(START, "parallel_invoke")→graph.add_edge("parallel_invoke", END)。这一步不能省略箭头符号，LangGraph会因缺少END边而拒绝编译图实例。
第四步：编译并启用流式输出app = graph.compile(stream_mode="values")。若不加stream_mode参数，批量执行时将等待全部Atom完成才返回结果，失去实时性优势。
执行批量任务
构造输入：必须是字典，且含"batch_inputs"键，值为长度≥2的列表，例如{"batch_inputs": ["query1", "query2", "query3"]}。
调用执行：for output in app.stream({"batch_inputs": batch_list}): print(output)。每次迭代返回一个字典，形如{"model_a": {...}, "model_b": {...}}，顺序与RunnableParallel中键声明顺序严格一致。
这一步操作起来很简单，直接把文件拖进去就行。
