解析 ‘Agentic Digital Twins’：如何利用 LangGraph 实时模拟一个工厂、城市或组织的逻辑运行状态？

各位技术同仁，下午好！

今天，我们聚焦一个前沿且极具潜力的技术领域——“Agentic Digital Twins”，即智能体数字孪生。我们将深入探讨如何利用 LangGraph 这一强大的工具，来实时模拟一个工厂、城市乃至整个组织的逻辑运行状态。这不仅仅是数据的镜像，更是行为与决策的模拟。

1. 智能体数字孪生：从数据到决策

1.1 什么是数字孪生？

在深入智能体数字孪生之前，我们先快速回顾一下传统数字孪生的概念。数字孪生是物理实体或系统在数字世界中的虚拟映射。它通过传感器、物联网设备等手段，实时收集物理世界的数据，并在数字模型中进行更新。其核心价值在于：

• 实时监测： 随时了解物理实体的当前状态。
• 性能分析： 基于历史数据和实时数据，分析系统性能。
• 预测性维护： 预测潜在故障，提前进行维护。
• 优化操作： 发现并改进操作流程中的效率瓶颈。

然而，传统的数字孪生虽然擅长捕捉物理属性和数据流，但在模拟“决策”、“意图”和“复杂交互”方面存在局限。它们更多是被动地反映现实，而非主动地参与推理和行动。

1.2 智能体数字孪生：更进一步的演化

智能体数字孪生（Agentic Digital Twins）是对传统数字孪生的一次范式升级。它将人工智能领域中的“智能体”（Agent）概念融入数字孪生中。这里的“智能体”不仅仅是一个数据处理单元，而是一个具备以下核心能力的实体：

• 感知（Perception）： 能够从数字孪生环境中获取信息（例如，读取传感器数据、理解其他智能体的状态）。
• 推理（Reasoning）： 能够基于感知到的信息和预设的目标，进行逻辑判断、问题解决和规划。
• 决策（Decision-making）： 能够根据推理结果，选择合适的行动方案。
• 行动（Action）： 能够通过工具（Tools）或直接修改数字孪生环境来执行决策。
• 自主性（Autonomy）： 能够在一定程度上独立运行，无需持续的人类干预。
• 交互性（Interaction）： 能够与其他智能体、人类用户或外部系统进行沟通和协作。

智能体数字孪生的核心特征

特征	传统数字孪生	智能体数字孪生
核心能力	数据映射、状态监测、物理仿真	决策、规划、交互、自主行动、逻辑仿真
驱动方式	物理数据流、预设规则	智能体目标、环境感知、AI推理（LLM）
模拟维度	物理状态、性能指标	逻辑流程、行为模式、决策链、组织动态
主要价值	实时洞察、预测维护、物理优化	策略验证、情景分析、自主优化、复杂系统管理
交互性	数据查询、可视化	智能体间协作、工具调用、人类干预
复杂系统	擅长物理系统	擅长模拟组织、城市等社会-技术复杂系统

智能体数字孪生的目标是模拟物理实体或系统在逻辑层面的运行状态，包括其决策过程、行为模式以及与其他实体间的复杂交互。例如，在一个工厂中，它不仅模拟机器的物理状态，更模拟生产经理如何根据订单、库存和设备状态做出生产排程决策；在一个城市中，它模拟交通部门如何根据实时路况调整红绿灯，或者市民如何选择出行方式。

2. LangGraph：构建智能体网络的利器

要构建这种具备复杂决策和交互能力的智能体系统，我们需要一个强大的框架来管理智能体的状态、行为、工具使用以及它们之间的通信流。LangGraph 正是为此而生。

2.1 LangGraph 概述

LangGraph 是 LangChain 生态系统中的一个高级模块，它专注于使用图结构来构建有状态的、多智能体应用程序。与 LangChain 中常见的简单链（Chains）不同，LangGraph 允许我们定义：

• 状态（State）： 在整个图的不同节点之间共享和更新的数据。
• 节点（Nodes）： 图中的基本处理单元，可以是单个智能体、一个工具调用、一个数据处理函数，甚至是另一个子图。
• 边（Edges）： 定义了状态如何从一个节点流向另一个节点。LangGraph 特别支持条件边，这意味着流程可以根据当前状态或节点的输出动态地改变路径。
• 循环（Cycles）： LangGraph 最强大的特性之一是它能够自然地处理循环，这对于需要迭代、自我纠正或多轮交互的智能体系统至关重要。

2.2 LangGraph 在智能体数字孪生中的优势

为什么 LangGraph 如此适合构建智能体数字孪生？

1. 状态管理： 数字孪生需要维护一个复杂的、不断变化的系统状态。LangGraph 的状态机制完美契合，允许所有智能体共享和更新这个中心化的“数字孪生状态”。
2. 多智能体协作： 一个复杂的系统通常由多个不同职责的智能体组成。LangGraph 的图结构可以清晰地定义这些智能体及其之间的交互路径。
3. 决策流与反馈循环： 智能体的决策往往是迭代的，一个决策的结果可能会影响后续的决策，形成反馈循环。LangGraph 的循环图设计能够自然地表达这种复杂的决策流和自适应行为。
4. 工具使用： 智能体需要与“数字孪生世界”进行交互，例如读取数据、执行模拟操作。LangGraph 能够无缝集成 LangChain 的工具（Tools），让智能体能够“感知”和“行动”。
5. 可观测性与调试： 图结构提供了清晰的流程视图，有助于理解智能体的行为和调试复杂问题。

3. 核心概念：LangGraph 构建块

在着手构建之前，我们先深入理解 LangGraph 的几个核心构建块。

3.1 定义状态（State）

状态是整个 LangGraph 的核心。它是一个 Python 字典或 Pydantic 模型，存储了所有节点共享和更新的信息。在智能体数字孪生中，这个状态就是我们的“孪生体”的实时逻辑快照。

from typing import TypedDict, List, Dict, Any
from datetime import datetime

class FactoryState(TypedDict):
    """
    一个工厂数字孪生的状态。
    """
    timestamp: datetime  # 当前模拟时间
    raw_materials: int   # 原材料库存
    widgets_produced: int # 已生产的部件数量
    widgets_in_qc: int   # 正在质量检测的部件数量
    finished_goods: int  # 成品库存
    demand: int          # 当前订单需求
    production_line_status: str # 生产线状态 ('idle', 'running', 'maintenance')
    events_log: List[str] # 事件日志，记录重要的操作和决策
    agent_decisions: Dict[str, Any] # 记录各智能体的最新决策

3.2 定义工具（Tools）

工具是智能体与数字孪生环境交互的接口。它们是普通的 Python 函数，但需要通过 LangChain 的 tool 装饰器进行封装，以便 LLM 能够理解并调用它们。在我们的模拟中，这些工具将模拟工厂的实际操作。

from langchain_core.tools import tool

# 假设这些是与工厂物理系统交互的模拟函数
# 它们会修改FactoryState中的某些部分，但为了保持示例简洁，
# 我们让这些工具只返回操作结果，实际状态更新由后续的SimulatorNode处理。

@tool
def start_production(quantity: int) -> str:
    """
    启动生产线，尝试生产指定数量的部件。
    Args:
        quantity: 尝试生产的部件数量。
    Returns:
        生产操作的结果描述。
    """
    if quantity <= 0:
        return "生产数量必须大于0。"
    return f"尝试启动生产线，目标生产 {quantity} 个部件。"

@tool
def stop_production() -> str:
    """
    停止生产线。
    Returns:
        停止生产操作的结果描述。
    """
    return "生产线已停止。"

@tool
def ship_widgets(quantity: int) -> str:
    """
    从成品库存中发货指定数量的部件。
    Args:
        quantity: 尝试发货的部件数量。
    Returns:
        发货操作的结果描述。
    """
    if quantity <= 0:
        return "发货数量必须大于0。"
    return f"尝试从成品库存发货 {quantity} 个部件。"

@tool
def order_raw_materials(quantity: int) -> str:
    """
    订购指定数量的原材料。
    Args:
        quantity: 尝试订购的原材料数量。
    Returns:
        订购操作的结果描述。
    """
    if quantity <= 0:
        return "订购数量必须大于0。"
    return f"已向供应商订购 {quantity} 单位原材料。"

@tool
def perform_quality_check(widget_id: int) -> str:
    """
    对指定部件执行质量检测。
    Args:
        widget_id: 待检测部件的唯一ID。
    Returns:
        质量检测结果的描述。
    """
    # 模拟质量检测过程，假设有90%的通过率
    import random
    if random.random() < 0.9:
        return f"部件 {widget_id} 质量检测通过。"
    else:
        return f"部件 {widget_id} 质量检测失败，需返工或报废。"

# 将所有工具组合成一个列表，供智能体使用
factory_tools = [start_production, stop_production, ship_widgets, order_raw_materials, perform_quality_check]

3.3 定义智能体（Agents/Nodes）

LangGraph 中的节点可以是智能体（通常由 LLM 驱动），也可以是普通函数。智能体节点通常负责：

1. 接收状态： 获取当前的数字孪生状态。
2. 感知与推理： 基于状态和预设目标，利用 LLM 进行推理。
3. 决策与行动： 根据推理结果，决定调用哪个工具，以及工具的参数。
4. 更新状态： 返回其决策或行动结果，以便更新 LangGraph 的共享状态。

我们将使用 LangChain 的 create_tool_calling_agent 来快速构建能够调用工具的 LLM 智能体。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import AgentExecutor, create_tool_calling_agent

# 假设已经配置了 OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 1. 生产经理智能体
production_manager_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂生产经理。你的职责是根据当前订单需求、原材料库存、成品库存和生产线状态，
     决定是否启动或停止生产线，以及生产多少部件。
     目标是满足需求，同时避免库存积压和原材料短缺。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
production_manager_agent_runnable = create_tool_calling_agent(llm, factory_tools, production_manager_prompt)
production_manager_agent = AgentExecutor(agent=production_manager_agent_runnable, tools=factory_tools, verbose=True)

# 2. 库存经理智能体
inventory_manager_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂库存经理。你的职责是监控原材料库存和成品库存。
     当原材料不足时，你需要订购原材料。
     当成品库存能够满足需求时，你需要安排发货。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
inventory_manager_agent_runnable = create_tool_calling_agent(llm, factory_tools, inventory_manager_prompt)
inventory_manager_agent = AgentExecutor(agent=inventory_manager_agent_runnable, tools=factory_tools, verbose=True)

# 3. 质量控制智能体
quality_control_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂质量控制主管。你的职责是对生产线上下来的部件进行质量检测。
     你主要关注widgets_in_qc中的部件数量，并对它们进行检测。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
quality_control_agent_runnable = create_tool_calling_agent(llm, factory_tools, quality_control_prompt)
quality_control_agent = AgentExecutor(agent=quality_control_agent_runnable, tools=factory_tools, verbose=True)

# 4. 需求生成器 (非LLM智能体，模拟外部事件)
def generate_demand(state: FactoryState) -> FactoryState:
    """
    模拟生成新的订单需求。
    """
    import random
    new_demand = random.randint(5, 20)
    print(f"n--- 模拟事件: 生成新需求 {new_demand} ---")
    state['demand'] += new_demand
    state['events_log'].append(f"[{state['timestamp']}] 新订单需求: {new_demand}")
    return state

# 5. 模拟器节点 (非LLM智能体，根据智能体决策更新物理状态)
def simulator_node(state: FactoryState) -> FactoryState:
    """
    根据智能体的决策和时间流逝，更新工厂的物理状态。
    这是一个核心节点，模拟真实世界对智能体操作的响应。
    """
    print("n--- 模拟器运行中 ---")
    # 模拟时间流逝
    state['timestamp'] = datetime.now() # 每次模拟步进都更新时间

    # 处理生产线状态和生产
    if state['production_line_status'] == 'running':
        # 假设每次模拟步进生产10个部件
        produced_this_step = min(10, state['raw_materials']) # 受原材料限制
        if produced_this_step > 0:
            state['raw_materials'] -= produced_this_step
            state['widgets_in_qc'] += produced_this_step # 生产出的部件进入QC队列
            state['events_log'].append(f"[{state['timestamp']}] 生产线生产了 {produced_this_step} 个部件，进入QC。")
        else:
            state['production_line_status'] = 'idle' # 没有原材料就停止生产
            state['events_log'].append(f"[{state['timestamp']}] 原材料不足，生产线停止。")

    # 处理质量检测
    if state['widgets_in_qc'] > 0:
        # 假设QC每次处理5个部件
        qc_processed_count = min(5, state['widgets_in_qc'])
        for i in range(qc_processed_count):
            # 这里我们简化，实际中应该由QualityControlAgent调用perform_quality_check
            # 并在其结果中判断是成功还是失败。
            # 为了模拟器独立更新状态，我们在这里直接处理
            if random.random() < 0.9: # 90%通过率
                state['finished_goods'] += 1
                state['events_log'].append(f"[{state['timestamp']}] 一个部件通过QC，进入成品库存。")
            else:
                state['events_log'].append(f"[{state['timestamp']}] 一个部件QC失败，报废。")
            state['widgets_in_qc'] -= 1

    # 处理发货
    if state['demand'] > 0 and state['finished_goods'] > 0:
        ship_count = min(state['demand'], state['finished_goods'])
        state['finished_goods'] -= ship_count
        state['demand'] -= ship_count
        state['events_log'].append(f"[{state['timestamp']}] 发货 {ship_count} 个部件，满足部分需求。")

    # 打印当前状态摘要
    print(f"  原材料: {state['raw_materials']}, 待QC: {state['widgets_in_qc']}, 成品: {state['finished_goods']}, 需求: {state['demand']}, 生产线: {state['production_line_status']}")

    return state

# 代理执行器包装器，将AgentExecutor的输出转换为适合LangGraph状态更新的格式
def agent_node(agent_executor: AgentExecutor, name: str):
    def node_function(state: FactoryState):
        current_state_str = (
            f"时间: {state['timestamp']}n"
            f"原材料库存: {state['raw_materials']}n"
            f"已生产部件: {state['widgets_produced']}n"
            f"待QC部件: {state['widgets_in_qc']}n"
            f"成品库存: {state['finished_goods']}n"
            f"当前订单需求: {state['demand']}n"
            f"生产线状态: {state['production_line_status']}n"
            f"最近事件: {state['events_log'][-3:] if len(state['events_log']) > 3 else state['events_log']}n"
        )
        result = agent_executor.invoke({"current_state": current_state_str})

        # 记录智能体的决策
        state['agent_decisions'][name] = result
        state['events_log'].append(f"[{state['timestamp']}] {name} 做出决策: {result.get('output', str(result))}")

        return state
    return node_function

production_manager_node = agent_node(production_manager_agent, "ProductionManager")
inventory_manager_node = agent_node(inventory_manager_agent, "InventoryManager")
quality_control_node = agent_node(quality_control_agent, "QualityControl")

3.4 定义图结构（Graph Structure）

使用 StateGraph 定义节点和边。

from langgraph.graph import StateGraph, END

# 创建图
workflow = StateGraph(FactoryState)

# 添加节点
workflow.add_node("generate_demand", generate_demand)
workflow.add_node("production_manager", production_manager_node)
workflow.add_node("inventory_manager", inventory_manager_node)
workflow.add_node("quality_control", quality_control_node)
workflow.add_node("simulator", simulator_node) # 模拟器节点，用于更新物理状态

# 设置入口点
workflow.set_entry_point("generate_demand")

# 定义边和条件逻辑

# 1. 每次模拟步进，首先生成新需求
workflow.add_edge("generate_demand", "inventory_manager") # 需求生成后，库存经理首先感知

# 2. 库存经理的决策后，可能需要生产经理介入，也可能直接进行模拟更新
workflow.add_edge("inventory_manager", "production_manager") # 库存经理之后，生产经理评估生产

# 3. 生产经理决策后，进入质量控制流程
workflow.add_edge("production_manager", "quality_control")

# 4. 质量控制后，更新模拟状态
workflow.add_edge("quality_control", "simulator")

# 5. 模拟器更新状态后，可以再次循环到需求生成，或者根据某些条件结束
# 这里我们设计为循环，模拟持续运行的工厂
workflow.add_edge("simulator", "generate_demand") 

# 编译图
app = workflow.compile()

4. 实践：构建一个简化的工厂数字孪生

现在，我们将上述概念整合起来，构建一个能够实时模拟工厂逻辑运行状态的智能体数字孪生。

4.1 场景设定

我们的工厂生产“部件”（Widgets）。工厂有以下几个关键环节：

• 原材料仓库： 存储生产所需的原材料。
• 生产线： 将原材料转化为部件。
• 质量控制： 对生产出的部件进行检测。
• 成品仓库： 存储通过质量检测的部件。
• 订单需求： 外部不断产生对部件的需求。

4.2 智能体职责

• 需求生成器 (generate_demand): 模拟外部订单的到来。
• 生产经理 (production_manager): 根据当前需求、库存和原材料，决定是否启动/停止生产，以及生产数量。
• 库存经理 (inventory_manager): 监控原材料和成品库存，决定是否订购原材料或安排发货。
• 质量控制主管 (quality_control): 监控待检测部件队列，并执行质量检测。
• 模拟器 (simulator): 这是一个特殊的非LLM节点，它根据智能体的决策和时间流逝，更新工厂的“物理”状态（例如，消耗原材料、增加成品、处理QC）。

4.3 模拟流程概览

1. 需求生成 (generate_demand): 模拟新的订单需求。
2. 库存经理 (inventory_manager): 检查成品库存是否能满足需求，以及原材料是否充足。如果需要，会调用订购或发货工具。
3. 生产经理 (production_manager): 根据库存经理的反馈和当前需求，决定是否启动生产线，并调用生产工具。
4. 质量控制 (quality_control): 检查是否有待检测的部件，并调用质量检测工具。
5. 模拟器 (simulator): 根据所有智能体的决策和工厂的内在逻辑（如生产效率、QC通过率），更新整个工厂的物理状态（原材料减少、成品增加、需求减少等）。
6. 流程循环回到需求生成，开始下一个模拟周期。

4.4 完整代码示例

import os
from typing import TypedDict, List, Dict, Any
from datetime import datetime
import random

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END

# --- 1. 设置 OpenAI API Key ---
# 请确保您的环境变量中已设置 OPENAI_API_KEY
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" 
# 或者直接在这里设置，但生产环境推荐使用环境变量
if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("OPENAI_API_KEY 环境变量未设置。请设置您的OpenAI API密钥。")

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# --- 2. 定义工厂状态 ---
class FactoryState(TypedDict):
    timestamp: datetime
    raw_materials: int
    widgets_produced: int
    widgets_in_qc: int
    finished_goods: int
    demand: int
    production_line_status: str # 'idle', 'running', 'maintenance'
    events_log: List[str]
    agent_decisions: Dict[str, Any] # 记录各智能体的最新决策

# --- 3. 定义工具 ---
@tool
def start_production(quantity: int) -> str:
    """
    启动生产线，尝试生产指定数量的部件。
    Args:
        quantity: 尝试生产的部件数量。
    Returns:
        生产操作的结果描述。
    """
    if quantity <= 0:
        return "生产数量必须大于0。"
    return f"尝试启动生产线，目标生产 {quantity} 个部件。"

@tool
def stop_production() -> str:
    """
    停止生产线。
    Returns:
        停止生产操作的结果描述。
    """
    return "生产线已停止。"

@tool
def ship_widgets(quantity: int) -> str:
    """
    从成品库存中发货指定数量的部件。
    Args:
        quantity: 尝试发货的部件数量。
    Returns:
        发货操作的结果描述。
    """
    if quantity <= 0:
        return "发货数量必须大于0。"
    return f"尝试从成品库存发货 {quantity} 个部件。"

@tool
def order_raw_materials(quantity: int) -> str:
    """
    订购指定数量的原材料。
    Args:
        quantity: 尝试订购的原材料数量。
    Returns:
        订购操作的结果描述。
    """
    if quantity <= 0:
        return "订购数量必须大于0。"
    return f"已向供应商订购 {quantity} 单位原材料。"

@tool
def perform_quality_check(widget_id: int) -> str:
    """
    对指定部件执行质量检测。
    Args:
        widget_id: 待检测部件的唯一ID。
    Returns:
        质量检测结果的描述。
    """
    # 模拟质量检测过程，假设有90%的通过率
    if random.random() < 0.9:
        return f"部件 {widget_id} 质量检测通过。"
    else:
        return f"部件 {widget_id} 质量检测失败，需返工或报废。"

factory_tools = [start_production, stop_production, ship_widgets, order_raw_materials, perform_quality_check]

# --- 4. 定义智能体节点和非LLM节点 ---

# 代理执行器包装器，将AgentExecutor的输出转换为适合LangGraph状态更新的格式
def agent_node_wrapper(agent_executor: AgentExecutor, name: str):
    def node_function(state: FactoryState):
        current_state_str = (
            f"时间: {state['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}n"
            f"原材料库存: {state['raw_materials']}n"
            f"已生产部件: {state['widgets_produced']}n"
            f"待QC部件: {state['widgets_in_qc']}n"
            f"成品库存: {state['finished_goods']}n"
            f"当前订单需求: {state['demand']}n"
            f"生产线状态: {state['production_line_status']}n"
            f"最近事件: {state['events_log'][-3:] if len(state['events_log']) > 3 else state['events_log']}n"
            f"重要提示: 请务必仅使用提供的工具函数进行操作。不要尝试直接修改工厂状态。"
        )
        print(f"n--- 智能体 {name} 运行中 ---")
        try:
            result = agent_executor.invoke({"current_state": current_state_str})
            # 记录智能体的决策或工具调用结果
            state['agent_decisions'][name] = result
            output_msg = result.get('output', str(result))
            state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] {name} 决策: {output_msg}")
            print(f"  {name} 决策结果: {output_msg}")
        except Exception as e:
            error_msg = f"智能体 {name} 执行失败: {e}"
            state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] {error_msg}")
            print(f"  {error_msg}")
            # 可以选择在这里抛出异常或返回特定状态来处理错误
            state['agent_decisions'][name] = {"error": error_msg}
        return state
    return node_function

# 生产经理智能体
production_manager_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂生产经理。你的职责是根据当前订单需求、原材料库存、成品库存和生产线状态，
     决定是否启动或停止生产线，以及生产多少部件。
     目标是满足需求，同时避免库存积压和原材料短缺。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
production_manager_agent_runnable = create_tool_calling_agent(llm, factory_tools, production_manager_prompt)
production_manager_agent_executor = AgentExecutor(agent=production_manager_agent_runnable, tools=factory_tools, verbose=False) # verbose=True for debugging LLM thoughts
production_manager_node = agent_node_wrapper(production_manager_agent_executor, "ProductionManager")

# 库存经理智能体
inventory_manager_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂库存经理。你的职责是监控原材料库存和成品库存。
     当原材料不足时，你需要订购原材料。
     当成品库存能够满足需求时，你需要安排发货。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
inventory_manager_agent_runnable = create_tool_calling_agent(llm, factory_tools, inventory_manager_prompt)
inventory_manager_agent_executor = AgentExecutor(agent=inventory_manager_agent_runnable, tools=factory_tools, verbose=False)
inventory_manager_node = agent_node_wrapper(inventory_manager_agent_executor, "InventoryManager")

# 质量控制智能体
quality_control_prompt = ChatPromptTemplate.from_messages([
    ("system", """
     你是一名工厂质量控制主管。你的职责是对生产线上下来的部件进行质量检测。
     你主要关注widgets_in_qc中的部件数量，并对它们进行检测。

     当前工厂状态:
     {current_state}

     请使用提供的工具来执行你的决策。
     """),
    ("placeholder", "{agent_scratchpad}"),
])
quality_control_agent_runnable = create_tool_calling_agent(llm, factory_tools, quality_control_prompt)
quality_control_agent_executor = AgentExecutor(agent=quality_control_agent_runnable, tools=factory_tools, verbose=False)
quality_control_node = agent_node_wrapper(quality_control_agent_executor, "QualityControl")

# 需求生成器 (非LLM智能体，模拟外部事件)
def generate_demand_node(state: FactoryState) -> FactoryState:
    import random
    new_demand = random.randint(5, 20)
    print(f"n--- 模拟事件: 生成新需求 {new_demand} ---")
    state['demand'] += new_demand
    state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 新订单需求: {new_demand}")
    return state

# 模拟器节点 (非LLM智能体，根据智能体决策和时间流逝更新物理状态)
def simulator_node_function(state: FactoryState) -> FactoryState:
    print("n--- 模拟器运行中 ---")
    state['timestamp'] = datetime.now() # 每次模拟步进都更新时间

    # 解析智能体决策并执行模拟操作
    # 这是一个关键步骤，将LLM智能体的“意图”转化为实际的状态变化

    # 1. 生产经理的决策
    pm_decision = state['agent_decisions'].get('ProductionManager', {})
    if pm_decision and 'tool_calls' in pm_decision:
        for tool_call in pm_decision['tool_calls']:
            if tool_call['name'] == 'start_production':
                target_quantity = tool_call['args'].get('quantity', 0)
                if state['production_line_status'] == 'idle' and target_quantity > 0:
                    state['production_line_status'] = 'running'
                    state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 生产线启动，目标生产 {target_quantity} 个部件。")
            elif tool_call['name'] == 'stop_production':
                if state['production_line_status'] == 'running':
                    state['production_line_status'] = 'idle'
                    state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 生产线停止。")

    # 2. 库存经理的决策
    im_decision = state['agent_decisions'].get('InventoryManager', {})
    if im_decision and 'tool_calls' in im_decision:
        for tool_call in im_decision['tool_calls']:
            if tool_call['name'] == 'order_raw_materials':
                order_quantity = tool_call['args'].get('quantity', 0)
                state['raw_materials'] += order_quantity # 假设订购立即到货
                state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 收到 {order_quantity} 单位原材料。")
            elif tool_call['name'] == 'ship_widgets':
                ship_quantity = tool_call['args'].get('quantity', 0)
                actual_ship = min(ship_quantity, state['finished_goods'])
                state['finished_goods'] -= actual_ship
                state['demand'] -= actual_ship # 满足部分需求
                state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 发货 {actual_ship} 个部件。")

    # 3. 质量控制主管的决策
    qc_decision = state['agent_decisions'].get('QualityControl', {})
    if qc_decision and 'tool_calls' in qc_decision:
        for tool_call in qc_decision['tool_calls']:
            if tool_call['name'] == 'perform_quality_check':
                # 质量检测的结果由工具本身模拟，这里只是更新状态
                # 假设每个QC操作处理一个部件
                if state['widgets_in_qc'] > 0:
                    state['widgets_in_qc'] -= 1
                    if random.random() < 0.9: # 90%通过率
                        state['finished_goods'] += 1
                        state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 一个部件通过QC。")
                    else:
                        state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 一个部件QC失败。")

    # 4. 模拟生产线实际运行
    if state['production_line_status'] == 'running':
        production_rate = 5 # 每次模拟步进生产5个部件
        can_produce = min(production_rate, state['raw_materials'])
        if can_produce > 0:
            state['raw_materials'] -= can_produce
            state['widgets_produced'] += can_produce # 累计生产总量
            state['widgets_in_qc'] += can_produce # 生产出的部件进入QC队列
            state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 生产线生产了 {can_produce} 个部件，进入QC。")
        else:
            state['production_line_status'] = 'idle' # 没有原材料就停止生产
            state['events_log'].append(f"[{state['timestamp'].strftime('%H:%M:%S')}] 原材料不足，生产线自动停止。")

    # 5. 打印当前状态摘要
    print(f"  当前状态: 原材料: {state['raw_materials']} | 待QC: {state['widgets_in_qc']} | 成品: {state['finished_goods']} | 需求: {state['demand']} | 生产线: {state['production_line_status']}")
    return state

# --- 5. 构建 LangGraph 工作流 ---
workflow = StateGraph(FactoryState)

workflow.add_node("generate_demand", generate_demand_node)
workflow.add_node("inventory_manager", inventory_manager_node)
workflow.add_node("production_manager", production_manager_node)
workflow.add_node("quality_control", quality_control_node)
workflow.add_node("simulator", simulator_node_function)

workflow.set_entry_point("generate_demand")

# 定义边：模拟流程的顺序
workflow.add_edge("generate_demand", "inventory_manager")
workflow.add_edge("inventory_manager", "production_manager")
workflow.add_edge("production_manager", "quality_control")
workflow.add_edge("quality_control", "simulator")
workflow.add_edge("simulator", "generate_demand") # 形成循环，持续模拟

app = workflow.compile()

# --- 6. 运行模拟 ---
initial_state: FactoryState = {
    "timestamp": datetime.now(),
    "raw_materials": 100,
    "widgets_produced": 0,
    "widgets_in_qc": 0,
    "finished_goods": 50,
    "demand": 0,
    "production_line_status": "idle",
    "events_log": ["工厂模拟开始"],
    "agent_decisions": {}
}

print("--- 工厂智能体数字孪生模拟启动 ---")
for i in range(10): # 运行10个模拟周期
    print(f"n======== 模拟周期 {i+1} ========")
    final_state = app.invoke(initial_state)
    initial_state = final_state # 将当前周期结束的状态作为下一个周期的初始状态

    # 打印一些关键指标
    print(f"n--- 周期 {i+1} 结束 ---")
    print(f"  总生产: {final_state['widgets_produced']}")
    print(f"  最终成品库存: {final_state['finished_goods']}")
    print(f"  最终原材料库存: {final_state['raw_materials']}")
    print(f"  剩余需求: {final_state['demand']}")

print("n--- 模拟结束 ---")
# 可以选择打印完整的事件日志
# for event in final_state['events_log']:
#     print(event)

代码解析：

1. 状态定义 (FactoryState): 包含了工厂的关键逻辑指标，如库存、需求、生产线状态等。
2. 工具定义 (@tool): 封装了工厂可以执行的原子操作，如 start_production, ship_widgets 等。智能体将通过调用这些工具来“操作”数字孪生。
3. 智能体节点 (production_manager_node, inventory_manager_node, quality_control_node):
   • 每个智能体都有一个特定的 Prompt，定义了其角色和目标。
   • 它们接收当前 FactoryState 的字符串表示，通过 LLM 进行推理。
   • LLM 的输出（通常是工具调用）会被捕获并记录在 agent_decisions 中。
4. 非LLM节点 (generate_demand_node, simulator_node_function):
   • generate_demand_node: 模拟外部事件，如新订单的到来。
   • simulator_node_function: 这是核心！ 它不是由 LLM 驱动的，而是根据智能体在当前周期做出的所有决策，以及工厂自身的物理/逻辑规则（如生产速率、QC通过率），来更新 FactoryState。它将智能体的“意图”转化为“现实”变化。
5. 图构建 (StateGraph):
   • add_node: 注册所有智能体和非LLM处理函数。
   • set_entry_point: 定义模拟的起始点。
   • add_edge: 定义了状态在节点间的流动顺序。这里形成了一个循环，使得模拟可以持续进行。
6. 运行模拟 (app.invoke): 循环调用 app.invoke，每个调用代表一个模拟周期。每个周期结束后的状态会作为下一个周期的输入，实现了状态的持续演进。

这个示例展示了如何将 LLM 驱动的智能体（负责决策）与传统的模拟逻辑（负责状态更新和规则执行）结合起来，形成一个动态、有反馈的智能体数字孪生。

5. 扩展到更复杂的场景：城市与组织

上述工厂的例子是一个起点，智能体数字孪生的威力在更宏大、更复杂的系统（如城市、大型组织）中才能得到充分体现。

5.1 城市数字孪生

想象一个城市的智能体数字孪生，它将包含：

• 状态: 交通流量、能源消耗、空气质量、人口分布、事件发生（事故、犯罪）、公共设施（医院、学校）容量、市民情绪指标等。
• 智能体:
  • 交通管理智能体: 根据实时交通流、天气预报，调整红绿灯配时、发布交通预警。
  • 能源管理智能体: 根据电力需求、可再生能源发电量，优化电网调度。
  • 应急服务智能体: 在发生事故时，调度警车、救护车、消防车，优化响应路径。
  • 城市规划智能体: 根据人口增长、资源消耗，评估新的基础设施建设方案。
  • 市民智能体 (群体): 模拟市民的出行、消费、互动行为，以预测城市活动模式。
• 工具: adjust_traffic_lights(intersection_id, schedule), allocate_power(area, amount), dispatch_emergency_unit(incident_id, unit_type), approve_building_permit(project_id).
• LangGraph 结构: 可以是分层的。例如，一个“城市协调者”智能体，下面管理着“交通”、“能源”、“应急”等子图，每个子图内部又包含多个智能体。

挑战与应对：

• 数据量巨大： 需要与实时传感器、数据库、地理信息系统（GIS）集成。利用向量数据库存储非结构化信息（如市民反馈），关系型数据库存储结构化数据。
• 计算复杂性： 模拟大量智能体的并行行为。可以考虑分布式计算、异步处理 LangGraph 节点。
• 涌现行为： 多个智能体交互可能产生意料之外的系统行为。这既是挑战也是价值所在，需要通过大量的模拟和分析来理解。

5.2 组织数字孪生

对一个大型组织进行逻辑模拟，其核心是模拟决策流程、资源分配和团队协作。

• 状态: 项目进度、预算分配、员工技能与负荷、市场趋势、销售线索、客户满意度、部门间沟通效率、企业文化指标等。
• 智能体:
  • 项目经理智能体: 根据项目目标、团队资源，制定任务计划、分配任务、监控进度。
  • 部门主管智能体: 管理部门成员的工作负荷、技能发展、绩效评估。
  • 销售智能体: 跟踪销售线索、制定销售策略、与客户互动。
  • 市场智能体: 分析市场数据、规划营销活动、评估品牌影响力。
  • HR智能体: 招聘、培训、管理员工福利、处理员工关系。
• 工具: assign_task(employee_id, task_id), approve_budget(department, amount), launch_marketing_campaign(details), hire_employee(role, skills), submit_report(report_type, data).
• LangGraph 结构: 同样可以设计为分层结构。例如，高层管理智能体可以设定季度目标，然后将目标分解给各个部门主管智能体，由他们进一步细化并分配给团队成员智能体。

挑战与应对：

• 抽象层级： 组织的逻辑比物理系统更抽象，需要更精细的Prompt工程来定义智能体的“价值观”和“行为准则”。
• 信息不对称： 模拟组织内部信息流的限制和偏差。智能体可以被设计为只能访问特定信息，模拟真实组织中的信息壁垒。
• 人类因素： 员工情绪、冲突、创新等难以量化的因素。可以通过引入情绪模型、随机事件或与人类反馈循环结合来部分解决。
• 验证与校准： 组织的“真实运行状态”往往没有精确的传感器数据。需要通过历史绩效数据、员工访谈、专家意见来校准模拟结果。

5.3 智能体数字孪生的普遍挑战

无论场景如何，构建智能体数字孪生都面临一些共性挑战：

• LLM的局限性： 幻觉、推理能力限制、成本和延迟。
  • 应对: 精心设计Prompt，使用更小的、专门优化的模型，实施缓存机制，结合传统算法进行关键决策。
• 状态复杂性管理： 随着系统规模扩大，状态会变得非常庞大且嵌套。
  • 应对: 使用 Pydantic 进行状态建模，将状态分解为多个子状态，利用数据库进行持久化和高效查询。
• 可解释性与调试： 智能体之间的复杂交互可能导致难以理解的系统行为。
  • 应对: 详尽的日志记录（如 events_log），可视化工具（如 LangGraph 的图可视化），以及设计智能体提供决策解释的机制。
• 验证与校准： 如何确保模拟结果的准确性和可靠性？
  • 应对: 与历史数据进行对比，进行A/B测试，引入领域专家进行人工评估，进行敏感性分析以理解参数变化的影响。
• 实时性与性能： “实时模拟”对响应速度有很高要求。
  • 应对: 异步执行，优化LLM调用，利用高性能计算资源，对非关键路径进行批量处理。
• 伦理与安全： 特别是在模拟人类行为和决策的场景中。
  • 应对: 确保智能体行为符合伦理规范，避免偏见，设计安全防护机制，引入人类监督和干预点。

6. 未来展望

智能体数字孪生技术正处于快速发展阶段。随着大型语言模型能力不断增强，以及 LangGraph 这类编排工具的成熟，我们有理由相信，这种技术将在以下方面带来变革：

• 更精准的预测与情景分析： 帮助决策者在复杂的现实世界中，预演不同策略的后果，选择最优路径。
• 自主化运营与优化： 在某些特定场景下，智能体数字孪生甚至可以与物理世界形成闭环，实现半自主或全自主的系统优化。
• 复杂系统设计与验证： 在系统投入实际运行前，通过数字孪生进行充分的测试和验证，降低风险。
• 教育与培训： 为各行各业的从业者提供高度仿真的训练环境。

从工厂车间的效率提升，到城市交通的智能调度，再到组织战略的优化，智能体数字孪生正在为我们开启一个全新的世界——一个不仅能看到数据，更能理解并预测行为和决策的世界。

7. 结语

智能体数字孪生，通过将传统数字孪生的数据洞察与智能体的自主决策能力相结合，为我们理解和优化复杂系统提供了一个革命性的框架。LangGraph 以其强大的图结构和状态管理机制，成为了构建这些智能体网络的理想工具。我们已经看到了其在工厂模拟中的潜力，而其在城市和组织等更宏大场景中的应用前景更是令人振奋。这项技术将赋能我们构建更具韧性、更高效、更智能的未来。