探讨 ‘Neural-Symbolic Hybrid Graphs’：在 LangGraph 中如何将确定性逻辑规则与概率性 LLM 推理完美融合

各位同仁，下午好！

今天，我们将深入探讨一个在人工智能领域日益受到关注，并且我认为是构建下一代智能系统的关键范式——“神经-符号混合图”（Neural-Symbolic Hybrid Graphs）。特别地，我们将聚焦于如何在 LangGraph 这一强大的框架中，将大型语言模型（LLMs）的概率性、涌现式推理能力，与传统符号系统的确定性、逻辑严谨性完美融合。这不仅仅是理论上的探索，更是实践中解决 LLM 局限性，构建更可靠、更可控、更可解释的智能应用的必由之路。

1. 神经与符号：两种智能范式的碰撞与互补

在深入混合图之前，我们首先需要理解神经（Neural）和符号（Symbolic）这两种人工智能范式的核心特点、优势与局限。

1.1 神经范式：LLMs 的崛起

近年来，以大型语言模型（LLMs）为代表的深度学习模型取得了惊人的进展。它们的核心优势在于：

• 强大的模式识别和泛化能力： LLMs 从海量数据中学习复杂的语言模式，能够处理模糊、非结构化的信息。
• 出色的自然语言理解与生成： 它们能够理解人类意图，生成流畅、富有创造性的文本。
• 涌现能力： 随着模型规模的增大，LLMs 展现出多步推理、情境感知等意想不到的能力。
• 适应性强： 通过微调或少样本学习，可以快速适应新的任务和领域。

然而，LLMs 并非万能，它们固有的一些局限性也日益凸显：

• 幻觉（Hallucination）： 模型可能生成听起来合理但实际上是错误或虚构的信息。
• 缺乏确定性与可解释性： 其内部工作机制是一个“黑箱”，很难追踪决策路径，也无法保证输出的绝对正确性。
• 逻辑推理与精确计算的弱点： 对于严格的逻辑推理、精确的数学计算或遵守复杂的业务规则，LLMs 常常力不从心。
• 知识时效性与可控性： 模型的知识截止日期固定，难以实时更新；同时，很难强制模型遵守特定的外部知识或事实。

1.2 符号范式：确定性与严谨的基石

与神经范式相对，传统的符号 AI 系统（如专家系统、知识图谱、规划器、规则引擎）则具有截然不同的特点：

• 确定性与可解释性： 决策过程基于明确的规则和逻辑，每一步都可追踪、可解释。
• 逻辑严谨性： 擅长执行复杂的逻辑推理、约束满足和精确计算。
• 知识可控与更新： 知识以结构化形式表示，易于管理、更新和验证。
• 精确性高： 在特定领域和任务中，能够提供高度准确的结果。

但符号系统也有其固有的挑战：

• 脆性（Brittleness）： 对输入的变化或噪声非常敏感，缺乏泛化能力。
• 知识获取瓶颈： 规则和知识需要人工编码，成本高昂且难以扩展。
• 难以处理模糊和非结构化数据： 不擅长理解自然语言的细微差别或从噪声中提取模式。
• 灵活性差： 适应新任务或新领域需要大量重构。

1.3 弥合鸿沟：神经-符号的必要性

显而易见，神经和符号范式各有千秋，也各有短板。神经-符号混合方法的核心思想，正是取长补短，将 LLMs 的感知、泛化和自然语言能力与符号系统的推理、精确性和可解释性结合起来。我们希望构建的智能系统，既能像 LLMs 那样灵活地理解世界，又能像符号系统那样严谨地执行任务。

2. 什么是神经-符号混合图？

神经-符号混合图，顾名思义，是一种计算图结构，其节点和边可以同时代表神经组件（如 LLM 调用、嵌入查询）和符号组件（如逻辑规则、函数调用、数据库查询、API 交互）。

2.1 “图”的意义：状态与流程的编排

为什么是“图”？在复杂的智能应用中，任务往往不是线性的，而是涉及多步骤、条件判断、循环和并行执行。图结构天然适合描述这种状态机或工作流。每个节点代表一个操作或一个决策点，而边则定义了数据流和控制流。

在神经-符号混合图中：

• 节点（Nodes） 可以是：
  • 神经节点： 调用 LLM 进行文本生成、分类、信息提取、意图识别等。
  • 符号节点： 执行确定的业务逻辑、数据库查询、API 调用、数据校验、数学计算、规则引擎判断等。
• 边（Edges） 可以是：
  • 确定性边： 基于明确的条件（如某个变量的值、某个函数的返回结果）进行路由。
  • 概率性/LLM 驱动的边： 由 LLM 的输出（如工具选择、下一步行动建议）来决定流程走向。

2.2 混合图的优势

通过这种融合，我们能够实现：

• 鲁棒性（Robustness）： 用确定性规则纠正 LLM 可能出现的“幻觉”或不准确之处。
• 可解释性（Explainability）： 关键决策由符号逻辑驱动时，路径清晰可查。
• 效率（Efficiency）： 仅在必要时调用昂贵的 LLM，将确定性任务交给高效的符号组件。
• 控制与安全（Control & Safety）： 通过符号规则强制遵守业务约束、安全策略。
• 接地性（Grounding）： 将 LLM 的输出与真实世界的知识库、数据库或系统操作连接起来。

3. LangGraph：混合图的理想编排层

LangGraph 是 LangChain 生态系统中的一个高级库，专门用于构建基于 LLM 的有状态、多步骤应用。它的核心思想是将整个应用建模为一个状态机，其中每个“节点”执行一个操作（调用 LLM、执行工具、运行 Python 函数），并通过“边”定义了这些操作之间的转换逻辑。

LangGraph 的架构天然支持神经-符号混合图的构建：

• 节点（Nodes）：
  • LLM 节点： 可以直接封装一个 Runnable 接口的 LLM 调用。
  • 工具节点： LangChain Tool 封装了任何 Python 函数，这些函数通常代表确定性的外部系统交互或业务逻辑。
  • 自定义 Python 函数节点： 任何 Python 函数都可以作为节点，实现复杂的条件逻辑、数据转换或规则判断。
• 边（Edges）：
  • 确定性边： 可以基于前一个节点的输出，通过一个简单的 Python 函数（conditional_edge）来决定下一个节点。
  • LLM 驱动的边： LLM 自身可以输出下一个要调用的工具名称或决策，LangGraph 可以据此路由。
• 状态（State）： LangGraph 的 StateGraph 允许定义一个共享的、可变的状态对象，所有节点都可以读取和修改它。这使得神经和符号组件能够无缝地共享上下文和传递信息。

通过 LangGraph，我们能够以声明式的方式定义复杂的混合工作流，将 LLM 的智能决策与确定性逻辑的精准执行融合在一个统一的框架中。

4. 构建模块：LangGraph 中的神经与符号组件

在 LangGraph 中，我们如何具体地定义和使用神经与符号组件呢？

4.1 神经节点

神经节点通常涉及与 LLM 的交互，包括：

• LLM 调用： 最直接的方式，用于生成文本、提取信息、分类等。
• RAG（Retrieval Augmented Generation）组件： 结合检索增强，让 LLM 能够访问外部知识库。
• 嵌入（Embeddings）与向量搜索： 用于语义相似度匹配。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable
from langchain_openai import ChatOpenAI # 或者其他LLM提供商
from typing import Dict, Any, List

# 定义一个简单的LLM节点
def create_llm_node(llm_model: ChatOpenAI, system_prompt: str) -> Runnable:
    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "{user_input}")
    ])
    return prompt | llm_model

# 示例：一个用于意图识别的LLM节点
llm = ChatOpenAI(model="gpt-4o", temperature=0)
intent_recognition_llm = create_llm_node(
    llm,
    "你是一个意图识别助手。根据用户的输入，判断他们的意图是 'query_product', 'place_order', 'check_status' 还是 'general_chat'。只返回意图名称。"
)

# 实际使用时，LLM节点通常还会结合Pydantic输出解析器，确保输出结构化
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser

class Intent(BaseModel):
    intent: str = Field(description="识别出的用户意图")

intent_parser = JsonOutputParser(pydantic_object=Intent)

intent_recognition_llm_with_parser = (
    ChatPromptTemplate.from_messages([
        ("system", "你是一个意图识别助手。根据用户的输入，判断他们的意图是 'query_product', 'place_order', 'check_status' 还是 'general_chat'。将结果输出为JSON格式，包含 'intent' 字段。")
        ("human", "{user_input}")
    ])
    | llm
    | intent_parser
)

# 示例调用
# result = intent_recognition_llm_with_parser.invoke({"user_input": "我想查一下iPhone 15的价格"})
# print(result) # {'intent': 'query_product'}

4.2 符号节点

符号节点是执行确定性逻辑的核心。它们可以是：

• 工具（Tools）： LangChain 的 Tool 封装了任何 Python 函数，通常用于与外部系统（数据库、API）交互或执行复杂计算。
• 纯 Python 函数： 直接作为 LangGraph 节点，用于实现业务规则、数据验证、状态转换等。
• 规则引擎： 更复杂的场景下，可以集成外部规则引擎。

import sqlite3
import json

# 模拟一个产品数据库
def get_db_connection():
    conn = sqlite3.connect(':memory:')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            category TEXT NOT NULL,
            price REAL NOT NULL,
            stock INTEGER NOT NULL
        )
    ''')
    products_data = [
        (1, 'iPhone 15', 'Electronics', 7999.00, 100),
        (2, 'MacBook Air', 'Electronics', 9999.00, 50),
        (3, 'Smartwatch X', 'Wearables', 1299.00, 200),
        (4, 'Wireless Earbuds', 'Audio', 899.00, 300),
        (5, 'Coffee Maker', 'Home Appliances', 599.00, 150)
    ]
    cursor.executemany('INSERT INTO products (id, name, category, price, stock) VALUES (?, ?, ?, ?, ?)', products_data)
    conn.commit()
    return conn

# 定义一个模拟数据库查询工具 (符号节点)
from langchain_core.tools import tool

@tool
def query_product_database(product_name: str = None, category: str = None) -> str:
    """
    查询产品数据库，根据产品名称或类别获取产品信息。
    Args:
        product_name (str): 要查询的产品名称。
        category (str): 要查询的产品类别。
    Returns:
        str: 包含产品信息的JSON字符串列表，如果未找到则为空列表。
    """
    conn = get_db_connection()
    cursor = conn.cursor()
    query = "SELECT name, category, price, stock FROM products WHERE 1=1"
    params = []
    if product_name:
        query += " AND name LIKE ?"
        params.append(f"%{product_name}%")
    if category:
        query += " AND category LIKE ?"
        params.append(f"%{category}%")

    cursor.execute(query, tuple(params))
    results = cursor.fetchall()
    conn.close()

    products_list = []
    for row in results:
        products_list.append({
            "name": row[0],
            "category": row[1],
            "price": row[2],
            "stock": row[3]
        })
    return json.dumps(products_list, ensure_ascii=False)

# 纯Python函数作为符号节点 (数据验证)
def validate_product_data(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    验证从数据库查询到的产品数据是否有效。
    如果产品列表为空或数据异常，则设置 'validation_status' 为 'failed'。
    """
    products_json = state.get('products_data', '[]')
    try:
        products = json.loads(products_json)
        if not products or not isinstance(products, list) or len(products) == 0:
            state['validation_status'] = 'failed'
            state['validation_message'] = '未找到匹配的产品或数据无效。'
        else:
            # 进一步检查产品数据的完整性和有效性（例如，价格不能为负）
            is_valid = all(p.get('name') and p.get('price', 0) > 0 for p in products)
            if not is_valid:
                state['validation_status'] = 'failed'
                state['validation_message'] = '产品数据存在异常。'
            else:
                state['validation_status'] = 'passed'
                state['validated_products'] = products # 存储解析后的产品对象
    except json.JSONDecodeError:
        state['validation_status'] = 'failed'
        state['validation_message'] = '数据库返回的产品数据格式错误。'
    return state

4.3 混合边（路由逻辑）

路由是 LangGraph 中融合神经和符号逻辑的关键。

• LLM 驱动的路由： LLM 的输出直接决定下一步的走向。例如，LLM 可能会输出一个工具名。
• 规则驱动的路由： 基于图状态中的某个变量值进行确定性判断。

# 示例：一个LLM决定路由的函数 (通常结合工具调用)
# LangGraph AgentExecutor 内部会处理LLM输出的Action/ToolCall，并自动路由到工具
# 但我们也可以显式地让LLM输出下一步的节点名称

class NextStep(BaseModel):
    next_node: str = Field(description="下一步要执行的节点名称")

next_step_parser = JsonOutputParser(pydantic_object=NextStep)

llm_router = (
    ChatPromptTemplate.from_messages([
        ("system", "根据当前查询的产品信息和验证状态，判断下一步应该怎么做。如果验证通过，返回 'generate_response'；如果验证失败，返回 'clarify_query'。请以JSON格式返回，包含 'next_node' 字段。")
        ("human", "当前产品信息: {products_data}n验证状态: {validation_status}n验证消息: {validation_message}")
    ])
    | llm
    | next_step_parser
)

# 示例：一个规则驱动的路由函数
def route_after_validation(state: Dict[str, Any]) -> str:
    """
    根据产品数据验证结果决定下一步路由。
    """
    if state.get('validation_status') == 'passed':
        return "generate_response"
    elif state.get('validation_status') == 'failed':
        return "clarify_query"
    else:
        return "error_handler" # 未知状态

5. 实践案例：构建神经-符号混合图

现在，让我们通过几个具体的例子来展示如何在 LangGraph 中构建神经-符号混合图。

5.1 案例一：产品信息查询与智能验证

场景描述： 用户询问产品信息。系统首先使用 LLM 提取产品名称或类别，然后调用一个符号工具（模拟数据库查询）获取数据。接着，一个符号函数验证查询结果的有效性。如果数据有效，LLM 生成响应；如果无效，LLM 会请求用户澄清，并可能尝试重新查询。

核心流程：
用户输入 -> LLM提取信息 (Neural) -> 查询数据库 (Symbolic Tool) -> 验证数据 (Symbolic Function) -> 条件路由 (Symbolic Rule) -> LLM生成响应 (Neural) 或 LLM请求澄清 (Neural) -> LLM提取信息 (Loop)

状态定义：
我们需要一个 TypedDict 来定义 LangGraph 的共享状态。

from typing import TypedDict, Annotated
import operator

# 定义图的状态
class ProductQueryState(TypedDict):
    user_input: str
    product_name: str # 从LLM提取
    category: str    # 从LLM提取
    products_data: str # 数据库返回的原始JSON字符串
    validation_status: str # 'passed', 'failed', 'pending'
    validation_message: str # 验证失败时的具体消息
    final_response: str # LLM生成的最终响应
    chat_history: List[Dict[str, str]] # 用于多轮对话

# 用于将新的信息合并到状态中
def update_state(current_state: ProductQueryState, new_values: Dict[str, Any]) -> ProductQueryState:
    current_state.update(new_values)
    return current_state

节点实现：

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnableLambda

# LLM for extracting product info
class ProductInfo(BaseModel):
    product_name: str = Field(description="用户想查询的产品名称，如果没有明确提及，则为空字符串")
    category: str = Field(description="用户想查询的产品类别，如果没有明确提及，则为空字符串")

product_info_parser = JsonOutputParser(pydantic_object=ProductInfo)

def extract_product_info(state: ProductQueryState) -> ProductQueryState:
    print("n--- 神经节点: 提取产品信息 ---")
    current_input = state["user_input"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个产品信息提取助手。从用户的输入中提取产品名称和类别。如果未明确提及，则留空。以JSON格式返回，包含 'product_name' 和 'category' 字段。"),
        ("human", "{user_input}")
    ])
    extraction_runnable = prompt | llm | product_info_parser
    extracted_data = extraction_runnable.invoke({"user_input": current_input})
    print(f"提取结果: {extracted_data}")
    return update_state(state, extracted_data)

# Symbolic tool for querying database
def call_product_db(state: ProductQueryState) -> ProductQueryState:
    print("n--- 符号节点: 调用产品数据库 ---")
    product_name = state.get("product_name")
    category = state.get("category")

    # 调用之前定义的 @tool 函数
    db_result_json = query_product_database.invoke({"product_name": product_name, "category": category})
    print(f"数据库查询结果: {db_result_json}")
    return update_state(state, {"products_data": db_result_json})

# Symbolic function for validating data (reusing the one defined earlier)
# def validate_product_data(state: ProductQueryState) -> ProductQueryState: ...

# LLM for generating final response
def generate_final_response(state: ProductQueryState) -> ProductQueryState:
    print("n--- 神经节点: 生成最终响应 ---")
    validated_products = state.get("validated_products", [])
    if validated_products:
        response_template = "我们找到以下产品信息：n"
        for p in validated_products:
            response_template += f"- {p['name']} ({p['category']}): 价格 {p['price']}元, 库存 {p['stock']}件。n"
        response_template += "n还有其他问题吗？"
    else:
        response_template = "抱歉，未能找到您查询的产品信息。"

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个友好的产品助手。根据提供的信息生成自然语言响应。"),
        ("human", response_template)
    ])
    final_response_text = (prompt | llm).invoke({"user_input": ""}) # LLM仅用于美化，主要内容已构建
    print(f"最终响应: {final_response_text.content}")
    return update_state(state, {"final_response": final_response_text.content})

# LLM for clarifying query
def clarify_query_llm(state: ProductQueryState) -> ProductQueryState:
    print("n--- 神经节点: 请求澄清 ---")
    validation_message = state.get("validation_message", "查询结果无效。")
    current_input = state["user_input"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"你是一个友好的产品助手。用户查询 '{current_input}' 得到了无效结果。原因可能是：{validation_message}。请引导用户提供更具体的信息，以便我们更好地帮助他们。"),
        ("human", "请用户澄清他的查询。")
    ])
    clarification_response = (prompt | llm).invoke({"user_input": ""})
    print(f"澄清请求: {clarification_response.content}")
    return update_state(state, {"final_response": clarification_response.content})

# 符号路由函数 (复用之前定义的 route_after_validation)
# def route_after_validation(state: ProductQueryState) -> str: ...

构建 LangGraph 图：

from langgraph.graph import StateGraph, END

# 创建一个StateGraph实例
workflow = StateGraph(ProductQueryState)

# 定义节点
workflow.add_node("extract_info", extract_product_info)
workflow.add_node("query_db", call_product_db)
workflow.add_node("validate_data", validate_product_data)
workflow.add_node("generate_response", generate_final_response)
workflow.add_node("clarify_query", clarify_query_llm)

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

# 添加边
workflow.add_edge("extract_info", "query_db")
workflow.add_edge("query_db", "validate_data")

# 添加条件边（符号路由）
workflow.add_conditional_edges(
    "validate_data",      # 从 validate_data 节点出来
    route_after_validation, # 路由函数
    {
        "generate_response": "generate_response", # 如果路由函数返回 "generate_response"，则到 generate_response 节点
        "clarify_query": "clarify_query"         # 如果路由函数返回 "clarify_query"，则到 clarify_query 节点
    }
)

# 最终响应和澄清请求都结束流程 (对于本轮对话)
workflow.add_edge("generate_response", END)
# 如果需要澄清，则回到提取信息阶段，等待用户新输入
workflow.add_edge("clarify_query", "extract_info") # 形成一个循环

# 编译图
app = workflow.compile()

# 运行示例
print("--- 开始产品查询流程 ---")
initial_state = ProductQueryState(
    user_input="我想知道iPhone 15的价格",
    product_name="", category="", products_data="",
    validation_status="pending", validation_message="", final_response="",
    chat_history=[]
)
final_output = app.invoke(initial_state)
print(f"n最终状态: {final_output}")
print(f"系统最终响应: {final_output['final_response']}")

print("n--- 尝试一个不存在的产品 ---")
initial_state_2 = ProductQueryState(
    user_input="有卖火星探测器吗？",
    product_name="", category="", products_data="",
    validation_status="pending", validation_message="", final_response="",
    chat_history=[]
)
final_output_2 = app.invoke(initial_state_2)
print(f"n最终状态: {final_output_2}")
print(f"系统最终响应: {final_output_2['final_response']}")

print("n--- 尝试一个无效查询 (产品名为空，类别也为空) ---")
initial_state_3 = ProductQueryState(
    user_input="给我推荐点东西",
    product_name="", category="", products_data="",
    validation_status="pending", validation_message="", final_response="",
    chat_history=[]
)
final_output_3 = app.invoke(initial_state_3)
print(f"n最终状态: {final_output_3}")
print(f"系统最终响应: {final_output_3['final_response']}")

解释：
这个例子清晰地展示了神经与符号的协同。LLM 负责理解模糊的自然语言（提取产品信息），而数据库查询和数据验证则是严格的符号操作。当符号层发现问题时（例如，未找到产品），流程会回到 LLM 驱动的澄清阶段，形成一个智能的纠错循环。

5.2 案例二：订单处理与业务规则校验

场景描述： 用户希望下订单。LLM 负责从对话中提取订单详情（商品、数量、收货地址）。然后，一系列符号节点会依次执行业务规则：检查库存、计算折扣、验证地址有效性。只有所有规则都通过，订单才能被“放置”（模拟操作）；否则，系统会告知用户哪个规则失败，并请求修正。

状态定义：

class OrderState(TypedDict):
    user_input: str
    item: str
    quantity: int
    address: str
    order_status: str # 'pending', 'processing', 'completed', 'failed'
    errors: List[str] # 存储遇到的错误信息
    final_message: str

节点实现：

# LLM for extracting order details
class OrderDetails(BaseModel):
    item: str = Field(description="用户想要订购的商品名称")
    quantity: int = Field(description="订购数量，默认为1")
    address: str = Field(description="收货地址")

order_details_parser = JsonOutputParser(pydantic_object=OrderDetails)

def extract_order_details(state: OrderState) -> OrderState:
    print("n--- 神经节点: 提取订单详情 ---")
    current_input = state["user_input"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个订单详情提取助手。从用户的输入中提取商品名称、数量和收货地址。数量默认为1。以JSON格式返回，包含 'item', 'quantity', 'address' 字段。"),
        ("human", "{user_input}")
    ])
    extraction_runnable = prompt | llm | order_details_parser
    extracted_data = extraction_runnable.invoke({"user_input": current_input})
    print(f"提取结果: {extracted_data}")
    return update_state(state, extracted_data)

# Symbolic node: Check stock
def check_stock(state: OrderState) -> OrderState:
    print("n--- 符号节点: 检查库存 ---")
    item = state.get("item")
    quantity = state.get("quantity", 0)
    errors = state.get("errors", [])

    # 模拟库存检查
    mock_stock = {"iPhone 15": 50, "MacBook Air": 20}
    if item not in mock_stock or mock_stock[item] < quantity:
        errors.append(f"库存不足：{item} 只有 {mock_stock.get(item, 0)} 件，您需要 {quantity} 件。")
        state["order_status"] = "failed"
    else:
        print(f"库存充足: {item} 尚有 {mock_stock[item]} 件。")

    state["errors"] = errors
    return state

# Symbolic node: Validate address
def validate_address(state: OrderState) -> OrderState:
    print("n--- 符号节点: 验证地址 ---")
    address = state.get("address")
    errors = state.get("errors", [])

    # 模拟地址验证（例如，检查是否包含“市”和“区”）
    if not address or "市" not in address or "区" not in address:
        errors.append(f"地址无效：'{address}' 格式不正确，请提供详细地址。")
        state["order_status"] = "failed"
    else:
        print(f"地址有效: {address}")

    state["errors"] = errors
    return state

# Symbolic node: Place order
def place_order(state: OrderState) -> OrderState:
    print("n--- 符号节点: 下订单 ---")
    item = state.get("item")
    quantity = state.get("quantity")
    address = state.get("address")

    # 模拟订单放置操作
    print(f"成功下订单: {quantity} 件 {item}，发往 {address}")
    state["order_status"] = "completed"
    state["final_message"] = f"您的 {quantity} 件 {item} 订单已成功提交，将发往 {address}。"
    return state

# LLM to inform user about errors
def inform_user_of_errors(state: OrderState) -> OrderState:
    print("n--- 神经节点: 告知用户错误 ---")
    errors = state.get("errors", [])
    error_message = "在处理您的订单时遇到以下问题：n" + "n".join(f"- {e}" for e in errors) + "n请修正您的订单信息。"

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个友好的订单助手。请礼貌地告知用户他们订单中的错误，并请求他们修正。"),
        ("human", error_message)
    ])
    response = (prompt | llm).invoke({"user_input": ""})
    state["final_message"] = response.content
    return state

# Symbolic routing based on order status
def route_order_processing(state: OrderState) -> str:
    if state.get("order_status") == "failed":
        return "inform_errors"
    return "place_order"

构建 LangGraph 图：

order_workflow = StateGraph(OrderState)

order_workflow.add_node("extract_details", extract_order_details)
order_workflow.add_node("check_stock", check_stock)
order_workflow.add_node("validate_address", validate_address)
order_workflow.add_node("place_order", place_order)
order_workflow.add_node("inform_errors", inform_user_of_errors)

order_workflow.set_entry_point("extract_details")

order_workflow.add_edge("extract_details", "check_stock")
order_workflow.add_edge("check_stock", "validate_address")

# 关键的符号条件路由
order_workflow.add_conditional_edges(
    "validate_address",
    route_order_processing, # 路由函数
    {
        "place_order": "place_order",
        "inform_errors": "inform_errors"
    }
)

order_workflow.add_edge("place_order", END)
order_workflow.add_edge("inform_errors", END) # 告知错误后结束当前流程，等待用户重新输入

order_app = order_workflow.compile()

# 运行示例
print("n--- 开始订单处理流程 (成功案例) ---")
order_input_success = OrderState(
    user_input="我想订购2件iPhone 15，寄到北京市海淀区中关村大街1号",
    item="", quantity=0, address="", order_status="pending", errors=[], final_message=""
)
order_output_success = order_app.invoke(order_input_success)
print(f"n最终状态: {order_output_success}")
print(f"系统最终消息: {order_output_success['final_message']}")

print("n--- 开始订单处理流程 (库存不足案例) ---")
order_input_stock_fail = OrderState(
    user_input="我要订购100件iPhone 15，寄到上海市浦东新区陆家嘴",
    item="", quantity=0, address="", order_status="pending", errors=[], final_message=""
)
order_output_stock_fail = order_app.invoke(order_input_stock_fail)
print(f"n最终状态: {order_output_stock_fail}")
print(f"系统最终消息: {order_output_stock_fail['final_message']}")

print("n--- 开始订单处理流程 (地址无效案例) ---")
order_input_address_fail = OrderState(
    user_input="我想订购1件MacBook Air，寄到我家",
    item="", quantity=0, address="", order_status="pending", errors=[], final_message=""
)
order_output_address_fail = order_app.invoke(order_input_address_fail)
print(f"n最终状态: {order_output_address_fail}")
print(f"系统最终消息: {order_output_address_fail['final_message']}")

解释：
这个例子展示了如何将多个符号规则串联起来，形成一个复杂的业务流程。LLM 负责初始的自然语言解析，但后续的库存、地址校验都是由确定性的 Python 函数完成。一旦任何一个符号规则失败，流程都会转向 LLM 驱动的错误通知，从而提供一个既智能又严谨的用户体验。

5.3 案例三：混合代理：数据分析与报告

场景描述： 用户要求分析数据。系统使用 LLM 判断用户意图是需要执行代码进行数据分析，还是仅仅需要一个自然语言的总结。如果需要代码执行，则调用一个符号工具（如 Python 解释器）。代码执行结果（无论是成功还是错误）会再由 LLM 进行总结或解释。

状态定义：

class DataAnalysisState(TypedDict):
    user_query: str
    action_type: str # 'execute_code', 'generate_summary'
    code_to_execute: str # LLM生成的代码
    code_output: str # 代码执行结果
    final_report: str # 最终报告
    analysis_errors: List[str] # 代码执行错误

节点实现：

from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.agents import AgentFinish
from langchain_core.tools import tool
import pandas as pd
import io

# 模拟一个Python解释器工具 (符号节点)
@tool
def python_interpreter(code: str) -> str:
    """
    执行Python代码并返回其输出。支持pandas进行数据操作。
    """
    print(f"n--- 符号节点: 执行Python代码 ---")
    try:
        # 创建一个沙箱环境
        exec_globals = {"pd": pd, "io": io}
        exec(code, exec_globals)

        # 捕获可能的输出
        output_buffer = io.StringIO()
        original_stdout = sys.stdout
        sys.stdout = output_buffer

        # 再次执行，这次捕获输出
        exec(code, exec_globals)
        sys.stdout = original_stdout # 恢复stdout
        output = output_buffer.getvalue()

        print(f"代码执行成功，输出:n{output}")
        return output if output else "代码执行完成，无显式输出。"
    except Exception as e:
        error_msg = f"代码执行错误: {str(e)}"
        print(error_msg)
        return error_msg

import sys

# LLM for deciding action
class ActionDecision(BaseModel):
    action: str = Field(description="决定下一步是 'execute_code' (执行代码) 还是 'generate_summary' (生成总结)")
    code: str = Field(description="如果决定执行代码，这里是生成的Python代码。")

action_decision_parser = JsonOutputParser(pydantic_object=ActionDecision)

def decide_action(state: DataAnalysisState) -> DataAnalysisState:
    print("n--- 神经节点: 决定行动 ---")
    user_query = state["user_query"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个数据分析助手。根据用户查询，决定是需要执行Python代码进行数据分析，还是直接生成一个总结。如果需要代码，请生成可执行的Python代码（可以使用pandas库），并将其放入 'code' 字段。如果不需要代码，将 'code' 字段留空。以JSON格式返回，包含 'action' 和 'code' 字段。"),
        ("human", "{user_query}")
    ])
    decision_runnable = prompt | llm | action_decision_parser
    decision = decision_runnable.invoke({"user_query": user_query})
    print(f"LLM决定: {decision}")
    return update_state(state, {"action_type": decision["action"], "code_to_execute": decision["code"]})

# Symbolic node: Execute code
def execute_code_node(state: DataAnalysisState) -> DataAnalysisState:
    print("n--- 符号节点: 执行代码 ---")
    code = state["code_to_execute"]
    output = python_interpreter.invoke({"code": code})

    analysis_errors = []
    if "代码执行错误" in output:
        analysis_errors.append(output)
        state["analysis_errors"] = analysis_errors

    return update_state(state, {"code_output": output})

# LLM for summarizing results or errors
def summarize_result(state: DataAnalysisState) -> DataAnalysisState:
    print("n--- 神经节点: 总结结果/错误 ---")
    user_query = state["user_query"]
    code_output = state.get("code_output", "")
    analysis_errors = state.get("analysis_errors", [])

    if analysis_errors:
        summary_prompt = f"用户请求: '{user_query}'。代码执行过程中出现错误: {analysis_errors[0]}。请向用户解释错误，并提供可能的修正建议。"
    elif code_output:
        summary_prompt = f"用户请求: '{user_query}'。代码执行结果: n{code_output}n请根据这些信息，生成一份简洁的自然语言报告。"
    else:
        summary_prompt = f"用户请求: '{user_query}'。请直接生成一份自然语言报告。"

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个数据分析报告助手。根据用户请求和分析结果，生成清晰、友好的报告。"),
        ("human", summary_prompt)
    ])
    final_report = (prompt | llm).invoke({"user_input": ""})
    print(f"最终报告: {final_report.content}")
    return update_state(state, {"final_report": final_report.content})

# Routing based on LLM's action decision
def route_data_analysis(state: DataAnalysisState) -> str:
    if state["action_type"] == "execute_code":
        return "execute_code"
    elif state["action_type"] == "generate_summary":
        return "summarize_result"
    else:
        return "error_handler" # Fallback

构建 LangGraph 图：

data_analysis_workflow = StateGraph(DataAnalysisState)

data_analysis_workflow.add_node("decide_action", decide_action)
data_analysis_workflow.add_node("execute_code", execute_code_node)
data_analysis_workflow.add_node("summarize_result", summarize_result)

data_analysis_workflow.set_entry_point("decide_action")

data_analysis_workflow.add_conditional_edges(
    "decide_action",
    route_data_analysis,
    {
        "execute_code": "execute_code",
        "generate_summary": "summarize_result"
    }
)

data_analysis_workflow.add_edge("execute_code", "summarize_result")
data_analysis_workflow.add_edge("summarize_result", END)

data_analysis_app = data_analysis_workflow.compile()

# 运行示例
print("n--- 开始数据分析流程 (代码执行) ---")
analysis_input_code = DataAnalysisState(
    user_query="计算列表 [10, 20, 30] 的平均值",
    action_type="", code_to_execute="", code_output="", final_report="", analysis_errors=[]
)
analysis_output_code = data_analysis_app.invoke(analysis_input_code)
print(f"n最终状态: {analysis_output_code}")
print(f"系统最终报告: {analysis_output_code['final_report']}")

print("n--- 开始数据分析流程 (仅总结) ---")
analysis_input_summary = DataAnalysisState(
    user_query="总结一下最近的市场趋势",
    action_type="", code_to_execute="", code_output="", final_report="", analysis_errors=[]
)
analysis_output_summary = data_analysis_app.invoke(analysis_input_summary)
print(f"n最终状态: {analysis_output_summary}")
print(f"系统最终报告: {analysis_output_summary['final_report']}")

print("n--- 开始数据分析流程 (代码错误) ---")
analysis_input_error = DataAnalysisState(
    user_query="运行一段有错误的Python代码：print(x / 0)",
    action_type="", code_to_execute="", code_output="", final_report="", analysis_errors=[]
)
analysis_output_error = data_analysis_app.invoke(analysis_input_error)
print(f"n最终状态: {analysis_output_error}")
print(f"系统最终报告: {analysis_output_error['final_report']}")

解释：
这个案例展示了一个更复杂的混合代理模式。LLM 不仅理解意图，甚至能够生成代码（神经组件）。而代码的执行和错误处理则由一个确定性的 Python 解释器（符号工具）完成。最后，LLM 再次介入，将复杂的代码输出或错误信息转化为用户友好的自然语言报告。这体现了神经组件在“理解”和“解释”上的优势，以及符号组件在“执行”和“确保正确性”上的不可替代性。

6. 高级考虑与最佳实践

构建神经-符号混合图并非一蹴而就，以下是一些值得注意的高级考虑和最佳实践：

• 状态设计： 精心设计 TypedDict 状态，确保所有节点都能访问所需信息，并清晰地定义每个字段的用途和生命周期。避免状态过于庞大或过于零碎。
• 错误处理与恢复： 预见到 LLM 可能的错误（幻觉、格式不正确）和符号组件的错误（API 调用失败、数据验证不通过）。在设计图时，为这些错误情况添加明确的错误处理节点和恢复路径，例如重试、请求澄清、回退到默认值或通知人工。
• 可观测性与调试： 使用 LangGraph 内置的 stream_events() 或 debug() 功能来跟踪图的执行路径、每个节点的输入输出。良好的日志记录和可视化工具对于调试复杂的混合图至关重要。
• 性能优化： LLM 调用通常是整个工作流中最耗时的部分。考虑缓存 LLM 结果、并行执行独立的 LLM 调用或符号操作，以及仅在必要时才调用 LLM。
• 安全性： 当符号节点涉及执行外部代码（如案例三中的 Python 解释器）或访问敏感系统时，务必实施严格的沙箱机制和权限控制，防止恶意代码注入或数据泄露。
• 人机协作（Human-in-the-Loop）： 对于关键决策点，特别是涉及高风险操作（如金融交易、医疗诊断）或 LLM 信心不足时，可以设计节点让人工审核或介入。
• 测试策略： 分层测试。对符号节点（工具、规则函数）进行单元测试和集成测试，确保其确定性逻辑的正确性。对神经节点（LLM提示、解析器）进行少样本测试，验证其在不同输入下的表现。最后，对整个图进行端到端测试。

7. 神经-符号混合图的未来展望

神经-符号混合图代表了人工智能发展的一个重要方向。展望未来，我们可以预见以下趋势：

• 更深层次的融合： 不仅仅是简单地串联，而是让神经和符号组件在更细粒度上进行交互和协同，例如 LLM 能够动态生成或修改符号规则，或者符号系统能够为 LLM 提供结构化的知识图谱，增强其推理能力。
• 自适应与学习能力： 混合系统能够从交互中学习，自动优化其神经和符号组件之间的协调策略，甚至能够根据运行时的性能指标自动调整规则或提示。
• 更广泛的应用： 超越传统的客服和代理场景，混合图将在科学研究、医疗诊断、法律分析、工业控制等对准确性、可解释性和鲁棒性要求极高的领域展现巨大潜力。
• 与知识图谱的深度集成： 结合 LLM 的语义理解能力和知识图谱的结构化知识，构建能够进行复杂问答、推理和规划的强大系统。

结语

神经-符号混合图为我们提供了一个强大的框架，它巧妙地融合了 LLM 的灵活性与传统符号系统的精确性。通过 LangGraph，我们能够以直观且可维护的方式构建这些混合系统，克服单一范式的局限，迈向更智能、更可靠、更可控的 AI 应用。这不仅仅是技术上的进步，更是我们对人工智能本质理解的一次飞跃。