解析 ‘Nodes’ 的物理执行：如何在 LangGraph 节点中混合使用同步阻塞与异步并发任务？

各位编程领域的同仁，下午好！

今天，我们将深入探讨一个在构建复杂智能代理时至关重要的话题：如何在 LangGraph 节点中优雅且高效地混合使用同步阻塞任务与异步并发任务。随着大型语言模型（LLM）和多智能体系统的兴起，LangGraph 作为一个强大的框架，为我们构建有状态、多步骤的代理提供了坚实的基础。然而，真实世界的应用场景往往是复杂的，我们的代理节点可能需要同时处理CPU密集型计算、调用传统阻塞式库，以及执行大量I/O密集型网络请求。如何在这种混合环境中保持系统的响应性与吞吐量，正是我们今天讲座的核心。

1. LangGraph 节点与执行模型概述

首先，让我们回顾一下 LangGraph 的基本概念。LangGraph 允许我们通过定义一系列“节点”（Nodes）和它们之间的“边”（Edges）来构建有向无环图（DAG）或循环图。每个节点本质上是一个Python可调用对象（函数或方法），它接收当前的代理状态作为输入，执行一些逻辑，并返回对状态的更新。

LangGraph 的核心优势在于其状态管理和循环执行能力。当一个图被 compile() 后，我们可以通过 invoke() (同步执行) 或 ainvoke() (异步执行) 方法来驱动代理。

从执行角度来看，LangGraph 内部是“异步优先”的。这意味着，如果你定义了一个 async def 的节点，LangGraph 会在其异步事件循环中调度它。如果你的节点是 def 函数，LangGraph 会将其视为同步任务执行。

节点的两种基本形式：

• 同步节点 (def 函数):

  def my_sync_node(state: dict) -> dict:
      # 执行一些同步操作
      updated_value = state.get("key", 0) + 1
      return {"key": updated_value}

  这种节点会阻塞当前执行线程，直到其所有操作完成。

• 异步节点 (async def 函数):

  import asyncio
  
  async def my_async_node(state: dict) -> dict:
      # 执行一些异步操作
      await asyncio.sleep(1) # 模拟异步 I/O
      updated_value = state.get("key", 0) + 1
      return {"key": updated_value}

  这种节点可以在等待 I/O 操作时释放控制权给事件循环，允许其他任务并发执行，从而提高整体效率。

问题提出：
当一个 LangGraph 节点 内部 需要同时执行阻塞式（同步）操作和非阻塞式（异步）操作时，我们该如何设计这个节点，以确保不会阻塞事件循环，同时又能充分利用Python的异步能力？这正是我们今天要解决的核心挑战。

2. 同步与异步任务的本质

为了更好地理解如何混合使用它们，我们首先需要清晰地认识同步与异步任务的根本区别。

2.1 同步 (Blocking) 任务

• 定义: 当一个任务开始执行时，它会完全占用当前的执行线程，直到任务完成或者遇到一个明确的等待指令（例如，等待另一个进程或线程的结果）。在此期间，该线程无法执行其他任何工作。
• 关键词: def 函数，顺序执行。
• 适用场景:
  • CPU 密集型计算: 例如，复杂的数学运算、数据转换、图像处理、机器学习模型本地推理等。这些任务主要消耗CPU资源，即使是多线程也可能受限于全局解释器锁（GIL），但在单线程内，它们会完全占用CPU。
  • 调用阻塞式 I/O 库: 例如，一些旧的数据库驱动、文件系统操作 (open(), read(), write())、以及一些第三方库（如 requests 库，默认是阻塞的）没有提供异步接口。
• 缺点: 当遇到 I/O 密集型操作时，如果线程被阻塞在等待数据返回（网络延迟、磁盘读写）上，CPU 实际上是空闲的，但由于线程被占用，其他任务无法在该线程上执行，导致资源浪费和响应迟缓。

2.2 异步 (Non-Blocking) 任务

• 定义: 异步任务利用一个事件循环（Event Loop）来管理多个操作。当一个异步任务发起一个 I/O 操作（如网络请求）时，它不会等待操作完成，而是将控制权交还给事件循环。事件循环可以切换到执行其他任务。当 I/O 操作完成后，事件循环会通知原任务，并恢复其执行。
• 关键词: async def 函数，await 关键字，事件循环，并发。
• 适用场景:
  • I/O 密集型操作: 绝大多数网络请求（HTTP API 调用、WebSockets）、数据库查询（使用异步驱动）、文件系统操作（使用 aiofiles 等异步库）、消息队列等。这些任务大部分时间都在等待外部资源的响应。
  • 高并发服务: Web 服务器、API 网关等需要同时处理大量客户端连接的场景。
• 优点: 在等待 I/O 时不阻塞线程，允许单个线程高效地处理大量并发 I/O 操作，极大地提高了系统的吞吐量和响应速度。
• 缺点: 不适合 CPU 密集型任务，因为 CPU 密集型任务会完全占用事件循环，导致其他异步任务无法获得执行机会。

2.3 总结对比

特性	同步任务 (Sync)	异步任务 (Async)
函数定义	def func(...)	async def func(...)
I/O 处理	阻塞式：等待 I/O 完成，线程空闲	非阻塞式：发起 I/O 后交出控制权，线程可执行其他任务
并发模型	通过多线程或多进程实现（共享内存或消息传递）	通过事件循环和协程（单线程协作式多任务）实现
典型场景	CPU 密集型计算，传统阻塞 I/O 库	I/O 密集型操作（网络请求，数据库），高并发服务
性能影响	I/O 密集时，资源利用率低，响应慢	I/O 密集时，资源利用率高，响应快
上下文切换	由操作系统调度线程	由事件循环调度协程（开销小）

3. 为什么要在 LangGraph 节点中混合使用？真实世界场景

理解了同步与异步的本质，我们就能更好地理解为什么在 LangGraph 节点中混合使用它们是如此重要：

1. 代理决策与工具调用:

   • 同步部分: 代理可能需要对从LLM获得的原始文本进行复杂的本地解析（例如，使用正则表达式、Pydantic模型验证）以提取工具名称和参数，或者执行一些基于当前状态的复杂逻辑判断。这些是CPU密集型或本地数据处理任务。
   • 异步部分: 代理在决定使用外部工具时，通常需要调用远程API（例如，天气服务、股票查询、数据库查询、另一个LLM服务）。这些都是典型的I/O密集型任务。
   • 混合需求: 一个 tool_executor 节点可能在执行工具前需要同步地格式化输入，然后异步地调用工具，最后再同步地解析工具的异步返回结果。

2. 数据处理管道:

   • 异步数据获取: 从多个外部数据源（如REST API、消息队列、分布式存储）并行地获取数据。
   • 同步数据转换/分析: 对获取到的数据进行复杂的统计分析、特征工程、数据清洗或机器学习模型预测。这些通常是CPU密集型任务。
   • 混合需求: 一个数据摄取节点可能需要异步地从多个来源拉取数据，然后同步地对这些数据进行预处理和校验，最后再异步地将处理结果写入另一个存储系统。

3. 遗留系统集成:

   • 在现有系统中，很多核心业务逻辑或底层库可能是同步阻塞的。当我们将这些系统逐步迁移到基于 LangGraph 的异步架构时，我们不能一步到位地重写所有代码。
   • 混合需求: 一个节点可能需要调用一个遗留的、只提供同步接口的客户端库（例如，一个老旧的 SOAP 客户端），同时又需要与其他现代的异步服务（如 asyncpg 数据库驱动）交互。

4. 优化资源利用率:

   • 通过将 I/O 阻塞任务异步化，我们可以防止事件循环空转，从而提高单个服务器的吞吐量。
   • 通过将 CPU 密集型任务从事件循环中剥离到单独的线程，我们可以避免阻塞整个异步系统，即使这些任务本身是阻塞的。

简而言之，真实世界的复杂性要求我们的代理节点既能高效地利用CPU进行计算，又能快速响应并并发处理外部I/O，这就迫使我们必须掌握在异步上下文中处理同步阻塞任务的技巧。

4. Python 异步编程核心机制：桥接同步与异步

Python 的 asyncio 库提供了几种关键机制来在同步和异步世界之间搭建桥梁。在 LangGraph 节点中，我们主要关注以下几个：

4.1 asyncio.to_thread() (Python 3.9+)

这是在异步上下文中运行同步阻塞代码的最主要和推荐的方法。

• 作用: 将一个同步的可调用对象（函数）放到一个单独的线程中执行，并返回一个可等待对象（awaitable）。当这个可等待对象被 await 时，它会等待被调用的同步函数在新线程中执行完成，然后将结果返回给调用者。
• 原理: asyncio.to_thread() 在内部使用 ThreadPoolExecutor 来管理一个线程池。当你调用 to_thread() 时，你的同步函数会被提交到这个线程池中执行。主事件循环不会被阻塞，因为它只是将任务“外包”了出去，然后继续处理其他协程。
• 适用场景: 在 async def 函数中调用任何会阻塞事件循环的同步代码（CPU密集型任务、阻塞式I/O库、time.sleep() 等）。

代码示例:

import asyncio
import time
import os

# 一个模拟的同步阻塞函数 (CPU密集型或阻塞I/O)
def sync_blocking_task(data: str) -> str:
    current_thread_id = os.getpid() # 获取进程ID，线程ID更难直接获取，但表示在不同上下文
    print(f"  [{asyncio.current_task().get_name()}] (Sync Task in Thread {current_thread_id}) Starting heavy calculation for '{data}'...")
    time.sleep(2) # 模拟耗时操作，会阻塞当前线程
    result = f"Processed '{data}' synchronously in thread {current_thread_id}"
    print(f"  [{asyncio.current_task().get_name()}] (Sync Task in Thread {current_thread_id}) Finished heavy calculation.")
    return result

# 一个模拟的异步 I/O 任务
async def async_io_task(query: str) -> str:
    print(f"  [{asyncio.current_task().get_name()}] (Async Task) Fetching external data for '{query}'...")
    await asyncio.sleep(1) # 模拟异步网络请求
    result = f"Fetched '{query}' asynchronously"
    print(f"  [{asyncio.current_task().get_name()}] (Async Task) Finished fetching external data.")
    return result

async def main_demonstration():
    print(f"[{asyncio.current_task().get_name()}] Main demonstration started.")

    # 1. 在异步函数中调用同步阻塞任务，但使用 asyncio.to_thread()
    print(f"n[{asyncio.current_task().get_name()}] --- Demo 1: Calling sync task via to_thread ---")
    start_time = time.monotonic()
    sync_future = asyncio.to_thread(sync_blocking_task, "input_A")
    # 注意：这里我们立即await了，所以它看起来是顺序执行的，但sync_blocking_task是在另一个线程中运行的
    result_sync = await sync_future
    end_time = time.monotonic()
    print(f"[{asyncio.current_task().get_name()}] Result from sync task: {result_sync}")
    print(f"[{asyncio.current_task().get_name()}] Time taken for sync task (via to_thread, awaited immediately): {end_time - start_time:.2f}s")

    # 2. 并发执行一个同步阻塞任务 (通过 to_thread) 和一个异步 I/O 任务
    print(f"n[{asyncio.current_task().get_name()}] --- Demo 2: Concurrent sync (via to_thread) and async ---")
    start_time = time.monotonic()
    task_sync = asyncio.to_thread(sync_blocking_task, "input_B")
    task_async = async_io_task("query_C")

    # 使用 asyncio.gather 并发等待两个任务
    results_sync_and_async = await asyncio.gather(task_sync, task_async)
    end_time = time.monotonic()

    print(f"[{asyncio.current_task().get_name()}] Results from concurrent run: {results_sync_and_async}")
    # 理论上，总时间应该接近两个任务中耗时最长的一个（2秒）
    print(f"[{asyncio.current_task().get_name()}] Time taken for concurrent tasks: {end_time - start_time:.2f}s")

    print(f"n[{asyncio.current_task().get_name()}] Main demonstration finished.")

if __name__ == "__main__":
    # 给主任务一个名称，方便跟踪
    asyncio.run(main_demonstration(), debug=True)

输出分析:
你会发现 sync_blocking_task 尽管执行了 time.sleep(2)，但 async_io_task 仍然能够并发执行。总运行时间接近 sync_blocking_task 的2秒，而不是 2 + 1 = 3 秒，这证明了并发性。asyncio.current_task().get_name() 会显示主协程名称，而 sync_blocking_task 内部的打印会显示它在另一个线程中执行的模拟信息。

4.2 asyncio.create_task() 和 asyncio.gather()

这两个函数是管理并发异步任务的核心。它们允许你在一个 async def 函数中同时启动多个异步操作，而不需要等待每个操作立即完成。

• asyncio.create_task(coro):

  • 作用: 将一个协程 (coro) 包装成一个 Task 对象，并将其提交给事件循环以供调度。它会立即返回 Task 对象，而不会等待协程完成。
  • 何时使用: 当你需要启动一个后台协程，并且稍后才需要它的结果，或者它是一个“fire-and-forget”的任务时。

• *`asyncio.gather(awaitables, return_exceptions=False)`:**

  • 作用: 并发地运行一个或多个可等待对象（协程或 Task），并等待它们全部完成。它返回一个列表，包含所有可等待对象的执行结果，顺序与输入顺序一致。
  • 何时使用: 当你需要同时发起多个异步 I/O 请求，并等待所有请求都完成后才继续下一步时。它是实现异步并发最常用的工具之一。

代码示例:
（asyncio.gather() 的示例已经包含在上面的 main_demonstration 中，这里再单独强调一下 create_task()）

import asyncio
import time

async def fetch_data_from_api(api_name: str, delay: int) -> str:
    print(f"  [{asyncio.current_task().get_name()}] (API: {api_name}) Starting fetch, will take {delay}s...")
    await asyncio.sleep(delay)
    result = f"Data from {api_name} after {delay}s"
    print(f"  [{asyncio.current_task().get_name()}] (API: {api_name}) Finished fetch.")
    return result

async def main_concurrent_async():
    print(f"[{asyncio.current_task().get_name()}] Main concurrent async started.")
    start_time = time.monotonic()

    # 使用 create_task 启动多个异步任务
    task1 = asyncio.create_task(fetch_data_from_api("API_X", 3), name="Task-API_X")
    task2 = asyncio.create_task(fetch_data_from_api("API_Y", 1), name="Task-API_Y")
    task3 = asyncio.create_task(fetch_data_from_api("API_Z", 2), name="Task-API_Z")

    print(f"[{asyncio.current_task().get_name()}] All tasks launched, now waiting...")

    # 使用 await 等待单个任务（按顺序等待会失去并发优势）
    # result1 = await task1
    # result2 = await task2
    # result3 = await task3

    # 更优：使用 asyncio.gather 等待所有任务，实现真正的并发等待
    results = await asyncio.gather(task1, task2, task3)

    end_time = time.monotonic()
    print(f"[{asyncio.current_task().get_name()}] All tasks completed. Results: {results}")
    # 总时间应该接近最长任务的执行时间 (3秒)
    print(f"[{asyncio.current_task().get_name()}] Total time taken: {end_time - start_time:.2f}s")

if __name__ == "__main__":
    asyncio.run(main_concurrent_async(), debug=True)

输出分析:
你会看到三个 fetch_data_from_api 任务几乎同时开始，并在它们各自的延迟后完成。总运行时间大约是3秒，而不是 3 + 1 + 2 = 6 秒，这再次体现了异步并发的效率。

4.3 nest_asyncio (外部库)

• 作用: 允许在已经运行的 asyncio 事件循环中再次运行 asyncio 事件循环（即嵌套事件循环）。
• 适用场景: 主要用于特殊环境，如 Jupyter notebooks、IPython shell 或某些 Web 框架中，这些环境可能已经在后台运行了一个事件循环，而用户代码又需要调用 asyncio.run()。
• 警告: 在生产环境中，尤其是在服务器应用中，应尽量避免使用 nest_asyncio。它会使得异步代码的执行逻辑变得复杂，并可能引入难以调试的问题。正确的设计通常是确保整个应用只有一个主事件循环，并在此循环中调度所有协程。对于在异步节点中调用阻塞代码，asyncio.to_thread() 是更安全、更推荐的方案。

5. 在 LangGraph 节点中实现混合执行

现在我们将上述 Python 异步机制应用到 LangGraph 节点的具体构建中。

5.1 场景一：异步节点调用同步阻塞任务 (推荐模式)

这是最常见且推荐的模式。如果你的 LangGraph 节点需要执行任何异步操作（如调用LLM、外部API），那么它应该被定义为 async def。当这个异步节点需要执行某个阻塞性任务时，就使用 asyncio.to_thread() 将其卸载到单独的线程中。

示例：一个智能体决策节点

假设我们有一个智能体，它需要：

1. 同步解析用户输入中的特定模式（CPU密集）。
2. 异步调用一个外部天气API（I/O密集）。
3. 异步调用一个金融API来获取股票价格（I/O密集）。
4. 同步分析天气和股票数据，做出决策（CPU密集）。

import asyncio
import time
import re
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END

# --- 辅助函数：模拟同步和异步任务 ---

# 模拟一个CPU密集型的同步解析任务
def parse_user_input_sync(text: str) -> dict:
    current_thread_name = asyncio.current_task().get_name() if asyncio.current_task() else "MainThread"
    print(f"  [{current_thread_name}] (Sync Parser) Starting to parse: '{text[:20]}'...")
    time.sleep(0.5) # 模拟解析耗时
    match_weather = re.search(r"weather in (w+)", text, re.IGNORECASE)
    match_stock = re.search(r"stock price for (w+)", text, re.IGNORECASE)
    parsed_info = {
        "location": match_weather.group(1) if match_weather else None,
        "stock_symbol": match_stock.group(1).upper() if match_stock else None,
    }
    print(f"  [{current_thread_name}] (Sync Parser) Finished parsing. Result: {parsed_info}")
    return parsed_info

# 模拟一个异步的外部天气API调用
async def fetch_weather_async(location: str) -> Optional[dict]:
    if not location:
        return None
    current_task_name = asyncio.current_task().get_name()
    print(f"  [{current_task_name}] (Async Weather API) Fetching weather for {location}...")
    await asyncio.sleep(2) # 模拟网络延迟
    weather_data = {"location": location, "temperature": "25°C", "condition": "Sunny"}
    print(f"  [{current_task_name}] (Async Weather API) Finished fetching weather for {location}.")
    return weather_data

# 模拟一个异步的外部股票API调用
async def fetch_stock_price_async(symbol: str) -> Optional[dict]:
    if not symbol:
        return None
    current_task_name = asyncio.current_task().get_name()
    print(f"  [{current_task_name}] (Async Stock API) Fetching stock price for {symbol}...")
    await asyncio.sleep(1.5) # 模拟网络延迟
    stock_data = {"symbol": symbol, "price": 150.75, "currency": "USD"}
    print(f"  [{current_task_name}] (Async Stock API) Finished fetching stock price for {symbol}.")
    return stock_data

# 模拟一个CPU密集型的同步决策分析任务
def analyze_data_and_decide_sync(parsed_input: dict, weather_data: Optional[dict], stock_data: Optional[dict]) -> str:
    current_thread_name = asyncio.current_task().get_name() if asyncio.current_task() else "MainThread"
    print(f"  [{current_thread_name}] (Sync Analyzer) Starting analysis...")
    time.sleep(0.8) # 模拟分析耗时
    decision = "No specific action."
    if weather_data and weather_data["condition"] == "Sunny":
        decision = f"Suggest outdoor activity in {weather_data['location']}."
    if stock_data and stock_data["price"] > 100:
        decision += f" Consider investing in {stock_data['symbol']}."
    print(f"  [{current_thread_name}] (Sync Analyzer) Finished analysis. Decision: {decision[:50]}...")
    return decision

# --- LangGraph 状态定义 ---

class AgentState(TypedDict):
    user_input: str
    parsed_info: Optional[dict]
    weather_info: Optional[dict]
    stock_info: Optional[dict]
    final_decision: Optional[str]

# --- LangGraph 混合执行节点 ---

async def mixed_agent_node(state: AgentState) -> AgentState:
    graph_task_name = asyncio.current_task().get_name()
    print(f"n[{graph_task_name}] Entering mixed_agent_node...")
    user_input = state["user_input"]

    # 1. 在单独线程中执行同步解析任务
    print(f"[{graph_task_name}] Scheduling sync input parsing...")
    parsed_info = await asyncio.to_thread(parse_user_input_sync, user_input)

    # 2. 并发调度异步 API 调用
    print(f"[{graph_task_name}] Scheduling async API calls...")
    weather_task = fetch_weather_async(parsed_info["location"])
    stock_task = fetch_stock_price_async(parsed_info["stock_symbol"])

    weather_info, stock_info = await asyncio.gather(weather_task, stock_task)

    # 3. 在单独线程中执行同步决策分析任务
    print(f"[{graph_task_name}] Scheduling sync decision analysis...")
    final_decision = await asyncio.to_thread(analyze_data_and_decide_sync, parsed_info, weather_info, stock_info)

    print(f"[{graph_task_name}] Exiting mixed_agent_node.")
    return {
        "parsed_info": parsed_info,
        "weather_info": weather_info,
        "stock_info": stock_info,
        "final_decision": final_decision
    }

# --- 构建 LangGraph ---

graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent_logic", mixed_agent_node)
graph_builder.set_entry_point("agent_logic")
graph_builder.add_edge("agent_logic", END)

app = graph_builder.compile()

# --- 运行 LangGraph ---

async def run_agent():
    print("--- Starting LangGraph Agent Simulation ---")
    initial_state = {"user_input": "Tell me the weather in London and stock price for GOOG",
                     "parsed_info": None, "weather_info": None, "stock_info": None, "final_decision": None}

    start_total_time = time.monotonic()
    final_state = await app.ainvoke(initial_state, config={"recursion_limit": 5}) # 使用 ainvoke 触发异步执行
    end_total_time = time.monotonic()

    print("n--- LangGraph Agent Simulation Complete ---")
    print(f"Final State: {final_state}")
    print(f"Total execution time: {end_total_time - start_total_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(run_agent(), debug=True)

代码分析：

• mixed_agent_node 被定义为 async def，因为它的核心操作是异步的（await）。
• parse_user_input_sync 和 analyze_data_and_decide_sync 是同步的CPU密集型任务，它们通过 await asyncio.to_thread(...) 被安全地从事件循环中剥离到单独的线程中执行。这保证了事件循环在这些CPU密集型任务执行时不会被阻塞。
• fetch_weather_async 和 fetch_stock_price_async 是异步的I/O密集型任务，它们通过 asyncio.gather() 被并发地调度和等待，最大限度地减少了等待时间。
• 整个节点的总执行时间将接近于最长的异步I/O任务（2秒）与最长的同步CPU任务（0.8秒）之和，再加上一些小的开销，而不是所有任务时间简单相加。这展示了 asyncio.to_thread 和 asyncio.gather 结合带来的高效并发。

5.2 场景二：同步节点需要触发异步操作 (不推荐，但有时需理解)

通常情况下，如果一个 LangGraph 节点需要执行异步操作，那么它就应该被定义为 async def。将异步操作强行塞入一个 def 同步节点，往往意味着架构设计上的妥协或错误。然而，为了完整性，我们讨论一下这种场景以及为什么它不被推荐。

在一个 def 同步函数中，你不能直接使用 await 关键字。如果你想在这里执行一个异步函数，你唯一的选择是手动启动并运行一个新的事件循环，或者在一个现有的事件循环上调度任务。

问题点：

• 阻塞主事件循环: 如果你的 LangGraph 整体是通过 ainvoke() 运行的（即在一个事件循环中），那么在其中一个同步节点内部调用 asyncio.run() 会启动一个新的、嵌套的事件循环。这个 asyncio.run() 调用会阻塞当前的线程，直到其内部的异步任务完成。这意味着，即使你的同步节点内部的异步任务是 I/O 密集型的，它仍然会阻塞 LangGraph 的主事件循环，丧失了异步带来的并发优势。
• RuntimeError: Event loop is already running: 如果 asyncio.run() 被调用时，当前的线程中已经有一个事件循环在运行，它会抛出 RuntimeError。这正是 nest_asyncio 尝试解决的问题，但如前所述，它有其自身的复杂性和风险。

示例（仅为说明问题，不推荐在生产LangGraph中使用）：

import asyncio
import time
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END

# 模拟一个异步外部 API 调用
async def mock_async_service_call(data: str) -> str:
    current_task_name = asyncio.current_task().get_name()
    print(f"  [{current_task_name}] (Mock Async Service) Making call with: {data}")
    await asyncio.sleep(1) # 模拟异步 I/O
    return f"Async result for '{data}'"

class AgentState(TypedDict):
    input: str
    processed_data: Optional[str]
    fetched_data: Optional[str]

# --- LangGraph 同步节点尝试调用异步操作 (不推荐!) ---
def problematic_sync_node(state: AgentState) -> AgentState:
    current_thread_name = asyncio.current_task().get_name() if asyncio.current_task() else "MainThread"
    print(f"n[{current_thread_name}] Entering problematic_sync_node...")
    input_data = state["input"]
    processed_data = f"Processed_{input_data}_sync"

    fetched_data = "Error: Async operation could not be performed."

    try:
        # 尝试获取当前事件循环，如果存在，说明我们已经在异步上下文中
        loop = asyncio.get_running_loop()
        print(f"[{current_thread_name}] WARNING: An event loop is already running. "
              "Calling asyncio.run() will raise RuntimeError or block the main loop.")
        # 在这种情况下，如果你真的想在同步函数中运行异步代码，
        # 并且不希望阻塞主事件循环，通常是无法直接做到的。
        # 最糟糕的办法是使用 `nest_asyncio.apply()` 然后 `loop.run_until_complete(...)`
        # 但这会导致代码难以理解和调试。
        # 正确的做法是：将此节点改为 `async def`，并使用 `await`。
        # 或者，如果异步操作可以被推迟，让下一个节点处理它。
        print(f"[{current_thread_name}] Action: Simulating failure or pushing async work to next stage.")
        fetched_data = f"Async call skipped in sync node due to active loop."

    except RuntimeError:
        # 如果没有事件循环在运行，说明这个同步节点是从一个纯同步的上下文调用的
        # 此时，调用 asyncio.run() 会创建一个新的、临时的事件循环
        print(f"[{current_thread_name}] No event loop running. Safely calling asyncio.run(). "
              "NOTE: This still blocks the current thread.")
        fetched_data = asyncio.run(mock_async_service_call(processed_data + "_async_payload"))
        print(f"[{current_thread_name}] asyncio.run() finished.")

    print(f"[{current_thread_name}] Exiting problematic_sync_node.")
    return {"processed_data": processed_data, "fetched_data": fetched_data}

# --- 构建 LangGraph ---

graph_builder = StateGraph(AgentState)
graph_builder.add_node("problematic_node", problematic_sync_node)
graph_builder.set_entry_point("problematic_node")
graph_builder.add_edge("problematic_node", END)

app_sync = graph_builder.compile()

# --- 运行 LangGraph ---

# 场景 A: 从异步上下文调用 LangGraph (使用 ainvoke)
async def run_agent_async_context():
    print("n--- Running LangGraph with problematic_sync_node from ASYNC context ---")
    initial_state = {"input": "async_invoke_request", "processed_data": None, "fetched_data": None}
    start_time = time.monotonic()
    final_state = await app_sync.ainvoke(initial_state) # LangGraph 自身运行在事件循环中
    end_time = time.monotonic()
    print("--- ASYNC Context LangGraph Complete ---")
    print(f"Final state: {final_state}")
    print(f"Total time: {end_time - start_time:.2f}s")

# 场景 B: 从同步上下文调用 LangGraph (使用 invoke)
def run_agent_sync_context():
    print("n--- Running LangGraph with problematic_sync_node from SYNC context ---")
    initial_state = {"input": "sync_invoke_request", "processed_data": None, "fetched_data": None}
    start_time = time.monotonic()
    final_state = app_sync.invoke(initial_state) # LangGraph 自身运行在同步模式下
    end_time = time.monotonic()
    print("--- SYNC Context LangGraph Complete ---")
    print(f"Final state: {final_state}")
    print(f"Total time: {end_time - start_time:.2f}s")

if __name__ == "__main__":
    # 运行场景 A: 这会展示 RuntimeError 或导致 async call skipped
    asyncio.run(run_agent_async_context())
    print("n" + "="*80 + "n")
    # 运行场景 B: 这会成功调用 asyncio.run()，但会阻塞整个线程
    run_agent_sync_context()

结论：
从 LangGraph 的角度看，如果一个节点需要执行异步操作，它就应该被定义为 async def。然后，在那个 async def 节点内部，使用 asyncio.to_thread() 来处理任何同步阻塞的子任务。这是最干净、最推荐、最符合 Python 异步编程范式的做法。

6. 最佳实践与架构考量

为了在 LangGraph 中高效地混合同步与异步任务，请遵循以下最佳实践：

1. 异步优先原则 (Async-First Mindset):
   如果你的代理涉及任何网络I/O、数据库查询或与外部服务（如LLM API）交互，那么你的 LangGraph 节点应该默认为 async def。这是利用 Python 异步能力的基石。

2. 明确区分任务类型:

   • I/O 密集型任务: 始终使用 async def 和 await 来处理，并利用 asyncio.gather() 实现并发。
   • CPU 密集型或阻塞式 I/O 任务: 在 async def 节点内部，通过 asyncio.to_thread() 将这些任务卸载到单独的线程池中执行。

3. 细化节点粒度:
   考虑将复杂的、混合任务的节点拆分成更小、更专注的节点。例如：

   • 一个节点专门负责异步数据获取。
   • 一个节点专门负责同步数据处理。
   • 一个节点专门负责异步结果存储。
     LangGraph 的状态传递机制使得这种拆分非常自然，也能提高代码的可读性、可测试性和复用性。

4. 错误处理:
   当使用 asyncio.to_thread() 时，如果在被调用的同步函数中发生异常，这个异常会在 await to_thread 返回的 future 时重新抛出。你需要像处理普通异步操作一样，使用 try...except 块来捕获和处理这些异常。

5. 资源管理与并发限制:

   • asyncio.to_thread() 默认使用一个 ThreadPoolExecutor。如果你需要更精细地控制线程池的大小或行为，可以使用 loop.run_in_executor() 并传入自定义的 ThreadPoolExecutor 实例。
   • 对于并发的异步 I/O 操作，如果并发量非常大，你可能需要考虑使用 asyncio.Semaphore 来限制同时进行的请求数量，以避免过载下游服务或耗尽本地资源。

6. 避免在 async def 中直接执行耗时同步代码:
   即使是很小的同步代码块，如果它足够慢，也可能阻塞事件循环，导致整个异步系统响应变慢。如果无法确定代码是否阻塞，请对其进行性能分析。

7. 日志与调试:
   在异步环境中，日志记录变得尤为重要。使用 asyncio.current_task().get_name() 或类似的机制在日志中包含当前任务的标识，有助于跟踪不同并发任务的执行流程。

7. 结语

在 LangGraph 框架中构建高性能、响应式的智能代理，离不开对同步与异步任务混合执行的精妙掌握。通过采纳异步优先的策略，并在 async def 节点中巧妙运用 asyncio.to_thread() 来处理阻塞型操作，辅以 asyncio.gather() 进行并发异步调用，我们能够充分发挥 Python 的并发潜力，构建出既能高效处理I/O密集型任务，又能从容应对CPU密集型计算的强大代理。这种灵活的执行模型，正是我们应对未来复杂AI应用挑战的关键所在。