Featured image of post Langchain-Graph langgraph实战教程
Agent

Langchain-Graph langgraph实战教程

基础知识

  1. 什么是 Langchain GRAPH？

Langchain GRAPH 是 Langchain 框架中用于处理和操作知识图谱（Knowledge Graphs）的模块。知识图谱是一种用节点（entities）和边（relationships）来表示知识的结构化方式。Langchain GRAPH 可以帮助你：

  • 构建知识图谱： 从文本、数据库或其他来源提取信息并构建知识图谱。
  • 查询知识图谱： 使用自然语言或结构化查询来检索知识图谱中的信息。
  • 知识图谱增强的问答： 将知识图谱与大型语言模型（LLM）结合，以提供更准确、更具上下文的答案。
  • 推理： 在知识图谱上进行推理，发现新的关系或知识。
  1. 核心概念
  • Graph Store (图存储): 用于存储和管理知识图谱数据的后端。常见的有 Neo4j, NebulaGraph, Kùzu 等。Langchain 也支持内存中的 NetworkXEntityGraph 用于快速原型设计。
  • Graph Document Loader (图文档加载器): 用于从不同来源（如文本文件、网页、数据库）加载数据并转换为图结构。
  • Graph Transformer (图转换器): 用于将加载的文档转换为图谱中的节点和关系。通常会利用 LLM 来识别实体和关系。
  • Graph Cypher QA Chain (图 Cypher 问答链): 允许你使用自然语言提问，该链会将问题转换为 Cypher (一种图查询语言，常用于 Neo4j 等图数据库)，然后在图上执行查询并返回结果。
  • Knowledge Graph Index (知识图谱索引): 用于将知识图谱集成到检索增强生成 (RAG) 流程中，使 LLM 能够利用图谱中的信息。
  1. 与阿里云百炼平台的模型结合

阿里云百炼平台提供了多种强大的大语言模型，我们可以将这些模型用于 Langchain GRAPH 中的以下任务：

  • 实体提取 (Entity Extraction): 从文本中识别出关键的实体（如人名、地名、组织机构名等）。
  • 关系提取 (Relationship Extraction): 识别实体之间的关系。
  • 自然语言理解 (NLU): 理解用户用自然语言提出的关于图谱的问题。
  • 答案生成 (Answer Generation): 基于从图谱中检索到的信息生成自然的答案。

代码实战： LangGraph 与 function call 构建智能天气查询助手教程

本教程将引导您了解如何使用 LangChain 和 LangGraph 构建一个能够理解自然语言并调用工具（在这里是查询天气）的智能应用。我们将详细解析代码中的每个关键部分。

核心目标： 创建一个应用，用户可以用自然语言提问特定城市的天气（目前仅支持上海和北京），应用能自动判断是否需要调用天气查询工具，获取信息后返回给用户，并能记住对话上下文。

代码概览： 代码定义了一个天气查询工具 get_weather_updates，使用阿里巴巴的 DashScope 平台上的 qwen-max 模型作为语言模型，并通过 LangGraph 构建了一个状态机来协调模型调用和工具执行。它还使用了 MemorySaver 来保持对话的记忆。

  1. 环境准备与依赖

    在运行代码之前，请确保：

    安装必要的库:

    pip install langchain langgraph langchain_openai langchain_core typing os

    设置环境变量 DASHSCOPE_API_KEY: 这是访问 DashScope 平台模型所必需的 API 密钥。代码中包含检查此环境变量的逻辑。

    import os
if not os.getenv("DASHSCOPE_API_KEY"):
    # os.environ["DASHSCOPE_API_KEY"] = "your_actual_dashscope_api_key" # 仅供测试时直接设置
    raise ValueError("DASHSCOPE_API_KEY environment variable not set. Please set it before running.")

    知识点:

    • os.getenv("DASHSCOPE_API_KEY"): 从环境变量中读取 API 密钥，这是推荐的安全做法，避免将密钥硬编码到代码中。

  2. 工具定义 (Tool Definition)

    from langchain_core.tools import tool

@tool
def get_weather_updates(query: str) -> str:
    """
    查询城市当前天气 (Query current weather for a city)
    Use this tool to find out the current weather for a given city.
    """
    print(f"--- Tool 'get_weather_updates' called with query: {query} ---")
    query_lower = query.lower()
    if "上海" in query_lower or 'shanghai' in query_lower:
        return "now is 30 celsius, foggy"
    elif "北京" in query_lower or 'beijing' in query_lower:
        return "now is 20 celsius, sunny"
    else:
        return f"Weather information for {query} not available with this tool. Only Shanghai and Beijing are supported."

tools = [get_weather_updates]

    知识点:

    • @tool 装饰器: 这是 LangChain 提供的一个便捷方式，用于将一个普通的 Python 函数转换为 LangChain 工具。LLM 可以被训练或提示来理解何时以及如何调用这个工具。
    • 函数签名与文档字符串 (Docstring):
      • query: str: 类型提示明确了输入参数的类型。
      • -> str: 类型提示明确了返回值的类型。
      • 文档字符串 (Docstring): 至关重要！LLM 会分析工具的名称和文档字符串来决定：
        1. 何时调用此工具：基于用户的输入和文档字符串中描述的工具能力（例如，“查询城市当前天气”）。
        2. 如何调用此工具：需要传递什么参数（例如，query 参数代表城市名称）。 因此，清晰、准确的文档字符串对于工具的正确使用至关重要。
    • 工具逻辑: 函数内部实现了简单的天气查询逻辑，目前只支持“上海”和“北京”。在实际应用中，这里可以是对接真实的天气 API。
    • tools = [get_weather_updates]: 将定义好的工具放入一个列表，后续会绑定到 LLM 或在 LangGraph 中使用。

  3. 模型初始化与工具绑定

    from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="qwen-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0
)

model_with_tools = model.bind_tools(tools)

    知识点:

    • ChatOpenAI: 虽然名为 ChatOpenAI，但通过指定 base_url 和 api_key，它可以配置为使用兼容 OpenAI API 接口的其他模型服务，如此处的 DashScope。
    • model="qwen-max": 指定使用的具体模型名称。
    • base_url: DashScope 提供的与 OpenAI 兼容的 API 端点。
    • api_key: DashScope 的 API 密钥。
    • temperature=0: temperature 参数控制模型输出的随机性。设置为 0 时，模型输出会更具确定性，对于需要精确调用工具的场景通常是好的选择。
    • model.bind_tools(tools): 这是一个关键步骤，它将我们定义的 tools 列表与 LLM 实例进行绑定。这使得 LLM 在处理输入时，能够“意识”到这些工具的存在，并在认为合适的时候决定调用它们。绑定后，LLM 的输出可能会包含特殊的“工具调用”指令。

  4. LangGraph 状态定义与图构建

    LangGraph 允许我们以图（Graph）的形式定义多步骤、可循环的智能体行为。图中的节点代表操作，边代表操作之间的转换逻辑。

    ** 4.1 状态定义 (State Definition)**

    from langchain_core.messages import HumanMessage, AIMessage, ToolMessage # Added ToolMessage
from langgraph.graph import StateGraph, MessagesState

# MessagesState 是一个特殊的 LangGraph 状态类型，它内部维护一个消息列表。
# 每次节点返回 {"messages": [...]} 时，这些消息会被追加到现有状态的消息列表中。
# class MessagesState(TypedDict):
#     messages: Annotated[list, add_messages]

    知识点:

    • MessagesState: 这是 LangGraph 提供的一个预定义的状态类型，非常适合构建聊天机器人或需要维护消息历史的应用。它本质上是一个字典，其中有一个键 messages，对应的值是一个消息列表。
    • HumanMessage, AIMessage, ToolMessage: LangChain 中定义的消息类型，分别代表用户的输入、AI 的回复以及工具执行的结果。ToolMessage 用于封装工具调用的输出，并包含一个 tool_call_id 以便将结果与特定的工具调用请求关联起来。

    **** 4.2 图节点 (Graph Nodes)

    from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools) # 创建一个工具节点，用于执行列表中的工具

def call_model(state: MessagesState): # 定义 Agent 节点逻辑
    messages = state["messages"]
    print(f"--- Calling model with {len(messages)} messages. Last message type: {type(messages[-1])} ---")
    response = model_with_tools.invoke(messages) # 调用绑定了工具的 LLM
    return {"messages": [response]} # 将 LLM 的响应（可能是普通消息或工具调用请求）添加到状态中

    知识点:

    • ToolNode(tools): LangGraph 的预构建节点，专门用于执行工具。当上一个节点（通常是 LLM）决定调用工具时，ToolNode 会接收到工具调用请求，执行相应的工具函数，并将工具的输出（包装在 ToolMessage 中）返回。
    • call_model(state: MessagesState): 这是我们自定义的“代理”或“Agent”节点。
      • 它接收当前的 MessagesState 作为输入。
      • model_with_tools.invoke(messages): 调用绑定了工具的 LLM。LLM 会接收到目前为止的所有消息历史。
        • 如果 LLM 认为可以直接回答，它会返回一个 AIMessage。
        • 如果 LLM 认为需要调用工具，它会返回一个包含 tool_calls 属性的 AIMessage。这个 tool_calls 列表包含了要调用的工具名称、参数以及一个唯一的 tool_call_id。
      • return {"messages": [response]}: 节点的输出必须符合状态的结构。这里，我们将 LLM 的响应封装成一个列表，并用键 messages 返回，MessagesState 会自动将这个新消息追加到总的消息列表中。

    **** 4.3 条件边逻辑 (Conditional Edge Logic)

    from typing import Literal
from langgraph.graph import END

def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    # 检查最后一条消息是否是 AI 消息并且包含了工具调用请求
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls and len(last_message.tool_calls) > 0:
        print(f"--- LLM decided to use tools: {last_message.tool_calls} ---")
        return "tools" # 如果有工具调用，则路由到 "tools" 节点
    print("--- LLM decided NOT to use tools, or no tool calls found. Ending or proceeding. ---")
    return END # 否则，结束流程 (或根据需要路由到其他节点)

    知识点:

    • should_continue(state: MessagesState): 这个函数定义了从 agent 节点出发后的路由逻辑。它检查状态中的最后一条消息（通常是 agent 节点产生的 LLM 响应）。
    • last_message.tool_calls: 如果 LLM 决定调用工具，其返回的 AIMessage 对象会有一个 tool_calls 属性。这是一个列表，因为 LLM 理论上可以一次请求调用多个工具（尽管在此示例中，qwen-max 和大多数模型一次只调用一个）。
    • Literal["tools", END]: Python 的类型提示，表示该函数要么返回字符串 “tools”，要么返回 LangGraph 的特殊标记 END。
    • 路由决策:
      • 如果检测到 tool_calls，函数返回 "tools"，意味着工作流应转到名为 "tools" 的节点（即我们定义的 tool_node）。
      • 否则，函数返回 END，意味着当前轮次的流程结束，最终结果已经由 LLM 生成。

    **** 4.4 构建图并编译

    workflow = StateGraph(MessagesState) # 初始化状态图，并指定状态类型
workflow.add_node("agent", call_model) # 添加 "agent" 节点
workflow.add_node("tools", tool_node)  # 添加 "tools" 节点

workflow.set_entry_point("agent") # 设置图的入口点为 "agent" 节点

# 添加条件边：从 "agent" 节点出发，根据 should_continue 函数的返回值决定下一个节点
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools", # 如果 should_continue 返回 "tools"，则去 "tools" 节点
        END: END          # 如果 should_continue 返回 END，则结束
    }
)
workflow.add_edge("tools", "agent") # 从 "tools" 节点处理完后，总是回到 "agent" 节点

# 内存与检查点
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver() # 使用内存存储检查点，用于保存对话状态

# 编译图，并加入检查点机制
app = workflow.compile(checkpointer=checkpointer)

    知识点:

    • StateGraph(MessagesState): 创建一个状态图实例，并明确告知它管理的状态类型是 MessagesState。
    • workflow.add_node("node_name", node_function_or_callable): 向图中添加节点。第一个参数是节点的唯一名称，第二个参数是该节点要执行的逻辑（函数或可调用对象）。
    • workflow.set_entry_point("agent"): 指定当图第一次被调用时，从哪个节点开始执行。
    • workflow.add_conditional_edges(...): 添加条件边。
      • 第一个参数是起始节点名称 ("agent")。
      • 第二个参数是条件函数 (should_continue)，它的返回值将决定路径。
      • 第三个参数是一个字典，映射条件函数的返回值到目标节点名称。
    • workflow.add_edge("tools", "agent"): 添加一条常规边。在工具执行完毕后 ("tools" 节点)，流程总是返回到 "agent" 节点，以便 LLM 可以处理工具的输出并生成最终回复。这是一个典型的 ReAct (Reasoning and Acting) 循环。
    • MemorySaver(): LangGraph 的检查点（checkpoint）机制之一。MemorySaver 将每个线程（对话）的状态保存在内存中。这使得对话可以具有上下文记忆。对于生产环境，可能会使用更持久的存储，如 Redis、数据库等。
    • app = workflow.compile(checkpointer=checkpointer): 编译图。编译过程将图的定义转换为一个可执行的应用。传入 checkpointer 使得应用能够保存和加载对话状态。

  5. 执行与交互

    # 使用字符串作为 thread_id，用于区分不同的对话线程
thread_id = "chat_thread_42"
config = {"configurable": {"thread_id": thread_id}}

print("\nInvoking for Shanghai...")
shanghai_input = {"messages": [HumanMessage(content="what's the weather in Shanghai?")]}
try:
    # 第一次调用，传入初始消息
    final_state_shanghai = app.invoke(shanghai_input, config=config)

    # 从最终状态中提取 AI 的回复
    if final_state_shanghai["messages"] and isinstance(final_state_shanghai["messages"][-1], AIMessage):
        result_shanghai = final_state_shanghai["messages"][-1].content
        print(f"Final response for Shanghai: {result_shanghai}")
    else:
        print(f"Unexpected final state for Shanghai: {final_state_shanghai['messages']}")

    print("\nInvoking for Beijing (same thread)...")
    beijing_input_message = HumanMessage(content="what's the weather in Beijing?")
    # 第二次调用，在同一个线程 (thread_id 相同)
    # LangGraph 的 MessagesState 会自动将新的 beijing_input_message 追加到此线程已有的消息历史中
    final_state_beijing = app.invoke({"messages": [beijing_input_message]}, config=config)

    if final_state_beijing["messages"] and isinstance(final_state_beijing["messages"][-1], AIMessage):
        result_beijing = final_state_beijing["messages"][-1].content
        print(f"Final response for Beijing: {result_beijing}")
    else:
        print(f"Unexpected final state for Beijing: {final_state_beijing['messages']}")

except Exception as e:
    print(f"An error occurred: {e}")
    import traceback
    traceback.print_exc()

    知识点:

    • thread_id: 这是一个关键概念，用于区分不同的对话会话。当与 checkpointer 一起使用时，具有相同 thread_id 的调用会共享相同的对话历史。
    • config = {"configurable": {"thread_id": thread_id}}: 在调用 app.invoke 时传入此配置，LangGraph 会根据 thread_id 加载或保存对应的对话状态。
    • app.invoke(input_messages, config=config): 执行编译好的 LangGraph 应用。
      • input_messages: 对于 MessagesState，输入应该是一个字典，其中 messages 键对应一个包含新消息的列表。
      • 第一次调用上海天气时，输入是 {"messages": [HumanMessage(content="what's the weather in Shanghai?")]}。
      • 第二次调用北京天气时，由于 MemorySaver 和相同的 thread_id，MessagesState 已经包含了上海的对话历史。我们只需传入新的用户消息 {"messages": [HumanMessage(content="what's the weather in Beijing?")]}。MessagesState 的特性是它会自动将新消息追加到现有消息列表中，所以 LLM 在处理北京的请求时，是能看到之前关于上海的对话的。
    • 结果提取: final_state_shanghai["messages"][-1].content 获取最后一条 AI 消息的内容。通常，图执行完毕后，最后一条消息是 AI 对用户问题的最终回复。

  6. 执行流程梳理

    当用户输入 “what’s the weather in Shanghai?” 时：

    1. 入口 (agent 节点):

      • call_model 函数被调用。
      • model_with_tools.invoke 接收到 [HumanMessage(content="what's the weather in Shanghai?")]。
      • LLM (qwen-max) 分析输入和绑定的 get_weather_updates 工具的文档字符串。它判断出需要调用此工具，并确定参数 query 应该是 “Shanghai”。
      • LLM 返回一个 AIMessage，其中包含 tool_calls 属性，例如：tool_calls=[ToolCall(name='get_weather_updates', args={'query': 'Shanghai'}, id='call_abc123')]。
      • call_model 返回 {"messages": [AIMessage_with_tool_call]}。

    2. 条件路由 (should_continue):

      • should_continue 函数检查最后一条消息。发现有 tool_calls。
      • 函数返回 "tools"。

    3. 工具执行 (tools 节点):

      • ToolNode 接收到 tool_calls。
      • 它查找名为 get_weather_updates 的工具，并使用参数 {'query': 'Shanghai'} 调用它。
      • get_weather_updates("Shanghai") 执行，返回字符串 "now is 30 celsius, foggy"。
      • ToolNode 将此结果包装成一个 ToolMessage，例如：ToolMessage(content="now is 30 celsius, foggy", tool_call_id='call_abc123')。
      • ToolNode 返回 {"messages": [ToolMessage_with_result]}。

    4. 返回 Agent (agent 节点):

      • 图的边配置为从 tools 节点返回到 agent 节点。
      • call_model 再次被调用。此时，state["messages"] 包含了：
        1. HumanMessage(content="what's the weather in Shanghai?")
        2. AIMessage(..., tool_calls=[...])
        3. ToolMessage(content="now is 30 celsius, foggy", tool_call_id='call_abc123')
      • model_with_tools.invoke 接收到这三条消息。
      • LLM 现在看到了原始问题、它自己调用工具的决定以及工具的执行结果。
      • 基于这些信息，LLM 生成一个自然的回复，例如：AIMessage(content="The current weather in Shanghai is 30 degrees Celsius and foggy.")。
      • call_model 返回 {"messages": [AIMessage_final_response]}。

    5. 条件路由 (should_continue):

      • should_continue 检查最后一条消息（最终的 AI 回复）。这条消息没有 tool_calls。
      • 函数返回 END。

    6. 结束:

      • 图的执行结束。app.invoke 返回最终的状态 final_state_shanghai。
      • 代码从 final_state_shanghai["messages"][-1].content 中提取并打印最终回复。
      • MemorySaver 会保存包含所有这四条消息的 MessagesState 到与 thread_id="chat_thread_42" 关联的内存中。

    当后续询问北京天气时，由于 thread_id 相同，call_model 初始调用时，state["messages"] 就会包含之前上海的四条消息，再加上新的 HumanMessage(content="what's the weather in Beijing?")，使得 LLM 具有了上下文感知能力。

  7. 关键知识点回顾

    • Tool Definition (@tool): 方便地将函数暴露给 LLM，文档字符串是关键。
    • Model Binding (.bind_tools()): 让 LLM 知道有哪些工具可用。
    • MessagesState: LangGraph 中用于管理对话消息历史的便捷状态。
    • StateGraph: 定义应用流程的核心，包括节点和边。
    • Nodes (agent, tools): 执行具体操作的单元。ToolNode 是预置的工具执行器。
    • Conditional Edges: 基于函数逻辑动态决定流程走向。
    • END: 特殊标记，表示图的一个执行路径结束。
    • Checkpointer (MemorySaver): 实现对话记忆的关键，通过 thread_id 管理不同对话的状态。
    • Message Types (HumanMessage, AIMessage, ToolMessage): LangChain 中标准化的消息对象，用于在不同组件间传递信息。ToolMessage 包含了 tool_call_id 以便将结果与请求关联。

代码实战：使用阿里云百炼模型构建和查询知识图谱

在这个实战教程中，我们将演示如何：

  1. 设置环境： 安装必要的库。
  2. 连接阿里云百炼模型： 使用 Langchain 提供的接口连接百炼平台的 LLM。
  3. 准备数据： 使用一些示例文本数据。
  4. 构建知识图谱： 从文本中提取实体和关系，并构建一个内存中的知识图谱 (使用 NetworkXEntityGraph 作为示例，实际生产中推荐使用更专业的图数据库)。
  5. 查询知识图谱： 使用自然语言进行提问。

步骤 1: 设置环境

首先，你需要安装 Langchain 和其他必要的库。

pip install langchain langchain-community langchain-experimental aiohttp neo4j networkx beautifulsoup4 tiktoken dashscope
  • langchain, langchain-community, langchain-experimental: Langchain 核心库。
  • aiohttp: 异步 HTTP 请求库，某些 Langchain 组件可能需要。
  • neo4j: 如果你要使用 Neo4j 作为图存储，则需要安装。
  • networkx: 用于在内存中创建和操作图。
  • beautifulsoup4: 用于解析 HTML/XML，常用于从网页加载数据。
  • tiktoken: OpenAI 的 tokenizer，Langchain 中常用。
  • dashscope: 阿里云百炼 SDK。

步骤 2: 连接阿里云百炼模型

你需要拥有阿里云账号并开通百炼大模型服务，获取 API Key (DASHSCOPE_API_KEY)。

import os
from langchain_community.llms import Tongyi
from langchain_community.graphs import NetworkXEntityGraph
from langchain.chains import GraphQAChain # 旧版本可能是 GraphCypherQAChain
from langchain.prompts import PromptTemplate
from langchain.indexes.graph import GraphIndexCreator # 旧版本可能是 GraphQAChain
from langchain.document_loaders import TextLoader # 示例使用文本加载器
from langchain_core.documents import Document

# 设置你的阿里云百炼 API Key
# 建议使用环境变量来管理 API Key
os.environ["DASHSCOPE_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # 替换为你的真实 API Key

# 初始化阿里云百炼模型
# 你可以选择不同的模型，例如 qwen-turbo, qwen-plus, qwen-max 等
# llm = Tongyi(model_name="qwen-turbo")
# 对于图构建中更复杂的实体和关系抽取，可能需要更高能力的模型
llm_for_graph_extraction = Tongyi(model_name="qwen-plus")
llm_for_qa = Tongyi(model_name="qwen-turbo") # 问答可以使用稍轻量的模型

print("阿里云百炼模型已初始化。")

步骤 3: 准备数据

我们使用一些简单的文本数据作为示例。你可以将其替换为更复杂的文档或从其他来源加载。

# 示例文本数据
document_content = """
张三是北京大学的教授，他研究人工智能领域。
李四是张三的学生，他就读于清华大学，学习计算机科学。
北京大学和清华大学是中国顶尖的大学。
人工智能是计算机科学的一个重要分支。
王五是百度的工程师，百度是一家位于北京的科技公司，专注于人工智能。
张三和王五在一次人工智能会议上认识。
"""

# 将文本内容包装成 Langchain Document 对象
docs = [Document(page_content=document_content)]

print("示例文本数据已准备。")

步骤 4: 构建知识图谱 (使用 NetworkXEntityGraph)

我们将使用 GraphIndexCreator (或者旧版本中的直接使用 LLM 进行提取并手动添加到 Graph) 从文本中提取实体和关系，并构建一个内存中的知识图谱。

from langchain.graphs.graph_document import GraphDocument, Node, Relationship
from langchain.chains.graph_qa.extract_triple import GraphTripleExtractionChain # 用于提取三元组

# 方法一：使用 GraphIndexCreator (较新的推荐方式)
# GraphIndexCreator 在内部会调用 LLM 进行实体和关系的提取
# 注意：GraphIndexCreator 的实现可能依赖于特定的 LLM 能力和提示工程

# graph_index_creator = GraphIndexCreator(llm=llm_for_graph_extraction) # 使用能力更强的模型进行抽取
# graph = graph_index_creator.graph_from_documents(docs)

# print("知识图谱 (使用 GraphIndexCreator) 构建完成。")
# print("节点数:", len(graph.get_schema()['nodes']))
# print("关系数:", len(graph.get_schema()['relationships']))
# print("图谱概要 (Schema):", graph.get_schema())


# 方法二：手动提取三元组并构建图 (更灵活，但更复杂)
# 这种方法可以让你更好地控制提取过程和自定义提示

# 定义节点和关系的类型 (可选，但有助于结构化)
# 这部分通常需要根据你的领域知识进行设计
allowed_nodes = ["Person", "Organization", "Location", "Field", "Concept", "Event"]
allowed_relationships = ["IS_A", "WORKS_AT", "STUDIES_AT", "LOCATED_IN", "FOCUSES_ON", "PART_OF", "MEETS_AT"]

# 创建一个空的 NetworkX 图
knowledge_graph = NetworkXEntityGraph()

# 使用 GraphTripleExtractionChain 提取三元组
# 你可能需要根据百炼模型的特点调整提示 (prompt)
# Langchain 默认的提示可能不是最优的
# 以下是一个简化的示例，实际中提示工程非常重要

# prompt_template_str = """
# 从以下文本中提取实体及其之间的关系。
# 仅提取文本中明确提到的实体和关系。
# 实体类型应尽可能具体，例如：Person, Organization, Location, Field, Concept, Event。
# 关系类型应描述实体间的联系，例如：IS_A, WORKS_AT, STUDIES_AT, LOCATED_IN, FOCUSES_ON, PART_OF, MEETS_AT。
# 以 (Head Entity, Relation, Tail Entity) 的格式输出三元组，每个三元组占一行。
# 如果没有找到三元组，则不输出任何内容。

# 示例：
# 文本: "爱因斯坦出生在德国。"
# 输出:
# (爱因斯坦, BORN_IN, 德国)

# 文本: "{text}"
# 输出:
# """
# extraction_prompt = PromptTemplate.from_template(prompt_template_str)

# triple_extractor = GraphTripleExtractionChain.from_llm(
# llm=llm_for_graph_extraction,
#     prompt=extraction_prompt
# )

# # 或者使用更通用的方式，让LLM直接生成图结构
# # 这通常需要更复杂的提示和后处理

# print("开始从文本中提取三元组...")
# extracted_knowledge = []
# for doc in docs:
# # 这种方式直接让LLM生成图谱结构，对于百炼模型可能需要特定的 Prompt
#     try:
# # 尝试让LLM直接生成 GraphDocument 格式
# # 这需要非常精巧的 Prompt Engineering，让模型理解输出的JSON结构
# # 这里我们简化处理，假设有一个函数能调用LLM并解析其输出为 GraphDocument
# # 实际中，你可能需要一个专门的 chain 来做这件事，该 chain 内部调用 LLM
# # 并将 LLM 的文本输出解析成 Node 和 Relationship 对象。

# # 简化的概念性代码：
#         response = llm_for_graph_extraction.invoke(
# f"""
# 分析以下文本，并将其转换为知识图谱。
# 节点应该包含 'id' (实体名称) 和 'type' (实体类型，例如：Person, Organization, Location, Field, Concept, Event)。
# 关系应该包含 'source' (源节点 id), 'target' (目标节点 id), 和 'type' (关系类型，例如：IS_A, WORKS_AT, STUDIES_AT, LOCATED_IN, FOCUSES_ON, PART_OF, MEETS_AT)。
# 请以JSON格式输出，包含一个 'nodes' 列表和一个 'relationships' 列表。

# 文本：
# {doc.page_content}

# JSON输出：
# """
# )
# # print("LLM原始输出:", response) # 调试用

# # 解析 LLM 的输出 (这里需要根据 LLM 的实际输出来编写解析逻辑)
# # 这是一个非常理想化的情况，通常LLM不会直接完美输出这个结构
# # 你可能需要使用 PydanticOutputParser 或类似的工具来确保输出格式
# # 这里我们手动构造一些示例 GraphDocument，模拟提取过程

# # 模拟提取结果：
#         nodes_data = [
#             Node(id="张三", type="Person"),
# Node(id="北京大学", type="Organization"),
# Node(id="人工智能", type="Field"),
# Node(id="李四", type="Person"),
# Node(id="清华大学", type="Organization"),
# Node(id="计算机科学", type="Field"),
# Node(id="中国", type="Location"),
# Node(id="王五", type="Person"),
# Node(id="百度", type="Organization"),
# Node(id="北京", type="Location"),
# Node(id="人工智能会议", type="Event")
# ]
#         rels_data = [
# Relationship(source=Node(id="张三", type="Person"), target=Node(id="北京大学", type="Organization"), type="WORKS_AT"),
# Relationship(source=Node(id="张三", type="Person"), target=Node(id="人工智能", type="Field"), type="STUDIES_FIELD"), # 或者 RESEARCHES
# Relationship(source=Node(id="李四", type="Person"), target=Node(id="张三", type="Person"), type="IS_STUDENT_OF"),
# Relationship(source=Node(id="李四", type="Person"), target=Node(id="清华大学", type="Organization"), type="STUDIES_AT"),
# Relationship(source=Node(id="李四", type="Person"), target=Node(id="计算机科学", type="Field"), type="STUDIES_FIELD"),
# Relationship(source=Node(id="北京大学", type="Organization"), target=Node(id="中国", type="Location"), type="LOCATED_IN"), # 假设关系
# Relationship(source=Node(id="清华大学", type="Organization"), target=Node(id="中国", type="Location"), type="LOCATED_IN"), # 假设关系
# Relationship(source=Node(id="人工智能", type="Field"), target=Node(id="计算机科学", type="Field"), type="IS_BRANCH_OF"),
# Relationship(source=Node(id="王五", type="Person"), target=Node(id="百度", type="Organization"), type="WORKS_AT"),
# Relationship(source=Node(id="百度", type="Organization"), target=Node(id="北京", type="Location"), type="LOCATED_IN"),
# Relationship(source=Node(id="百度", type="Organization"), target=Node(id="人工智能", type="Field"), type="FOCUSES_ON"),
# Relationship(source=Node(id="张三", type="Person"), target=Node(id="王五", type="Person"), type="MET_AT", properties={"event": "人工智能会议"}),
# ]
#         graph_doc = GraphDocument(nodes=nodes_data, relationships=rels_data, source=doc)
#         extracted_knowledge.append(graph_doc)
#         print(f"从文档中模拟提取了 {len(nodes_data)} 个节点和 {len(rels_data)} 个关系。")

#     except Exception as e:
# print(f"从文档提取三元组失败: {e}")
#         continue

# # 将提取的知识添加到图中
# for gd in extracted_knowledge:
#     knowledge_graph.add_graph_documents([gd])

# print("\n知识图谱 (手动提取并构建) 构建完成。")
# print("图谱 Schema:", knowledge_graph.schema)
# print("所有三元组 (Triples):")
# for triple in knowledge_graph.get_triples():
#     print(triple)

# --- 切换到更简单和推荐的 LLMGraphTransformer 方法 ---
from langchain_experimental.graph_transformers.llm import LLMGraphTransformer

# 使用 LLMGraphTransformer 进行图构建
# 这通常是更推荐的方式，因为它封装了复杂的提示工程
# 你可以传入 allowed_nodes 和 allowed_relationships 来指导提取过程
# 如果不传入，它会尝试自动识别

llm_transformer = LLMGraphTransformer(
    llm=llm_for_graph_extraction,
    allowed_nodes=allowed_nodes, # 可选
    allowed_relationships=allowed_relationships # 可选
    # prompt= # 你也可以自定义提示，但这通常比较复杂
)

print("开始使用 LLMGraphTransformer 从文档构建图谱...")
graph_documents = llm_transformer.convert_to_graph_documents(docs)
print(f"\n成功从文档转换了 {len(graph_documents)} 个 GraphDocument。")

# 将转换后的 GraphDocument 添加到 NetworkX 图中
# 创建一个新的 NetworkXEntityGraph 实例来存储最终的图
final_graph = NetworkXEntityGraph()
final_graph.add_graph_documents(graph_documents)

print("\n知识图谱 (使用 LLMGraphTransformer) 构建完成。")
print("图谱 Schema:", final_graph.schema) # final_graph.schema 可能不直接显示所有节点和关系，取决于其内部实现
print("所有提取的三元组 (Triples):")
for triple in final_graph.get_triples():
    print(triple)

# 你也可以直接查看 NetworkX 图的节点和边
print(f"\nNetworkX 图中的节点数: {final_graph.graph.number_of_nodes()}")
print(f"NetworkX 图中的边数: {final_graph.graph.number_of_edges()}")
# print("节点:", list(final_graph.graph.nodes(data=True))) # 打印节点及其属性
# print("边:", list(final_graph.graph.edges(data=True)))   # 打印边及其属性

重要提示关于图构建：

  • 提示工程 (Prompt Engineering): 使用 LLM 进行实体和关系提取时，提示的设计至关重要。你需要仔细设计提示，以便模型能够准确地识别出你感兴趣的实体类型和关系类型。对于百炼模型，你可能需要根据其特性进行特定的优化。
  • LLMGraphTransformer： 这是目前 Langchain 中推荐的用于从文档构建图谱的方式。它内部处理了与 LLM 的交互和提示。你可以通过 allowed_nodes 和 allowed_relationships 参数来指导提取过程，使其更符合你的需求。如果不提供这些参数，Transformer 会尝试自动推断。
  • 迭代和评估： 图构建通常是一个迭代的过程。你需要不断评估提取的质量，并根据需要调整提示或使用的模型。
  • PydanticOutputParser: 对于更复杂的提取任务，并希望LLM输出结构化的JSON，可以考虑使用 PydanticOutputParser 来定义期望的输出模式，并解析LLM的输出。
  • 错误处理和鲁棒性： 在实际应用中，LLM的输出可能并不总是完美的。你需要添加错误处理逻辑来处理不符合预期格式或不准确的提取结果。

步骤 5: 查询知识图谱

现在我们已经有了一个知识图谱，可以使用 GraphQAChain (或旧版本的 GraphCypherQAChain，但对于内存中的 NetworkXEntityGraph，通常是前者或直接的图遍历/查询) 来用自然语言提问。

# 对于 NetworkXEntityGraph，我们通常不直接使用 Cypher 查询。
# GraphQAChain 会尝试理解问题，并在图上查找相关信息，然后用LLM生成答案。

# 创建 GraphQAChain
# 注意：对于 NetworkXEntityGraph，GraphQAChain 的行为可能与基于 Cypher 的图数据库不同。
# 它更多的是利用图的结构进行上下文检索，然后让 LLM 回答。

graph_qa_chain = GraphQAChain.from_llm(
    llm=llm_for_qa,  # 用于生成答案的 LLM
    graph=final_graph, # 我们构建的图
    verbose=True  # 打印中间步骤，便于调试
)

# 提出问题
questions = [
    "张三是做什么工作的？",
    "谁是张三的学生？",
    "李四在哪里学习？",
    "百度公司专注于哪个领域？",
    "张三和王五是在哪里认识的？",
    "列出所有的人物和他们的职业或研究领域。",
    "北京大学和清华大学位于哪个国家？"
]

for question in questions:
    print(f"\n问题: {question}")
    try:
        answer = graph_qa_chain.run(question) # 或者使用 .invoke({"query": question})
        print(f"答案: {answer}")
    except Exception as e:
        print(f"回答问题时出错: {e}")
        # 对于 NetworkXEntityGraph，如果 GraphQAChain 内部尝试生成 Cypher，可能会出错。
        # 可以尝试直接从图中提取信息并让LLM总结。

# 如果 GraphQAChain 对于 NetworkXEntityGraph 支持不佳，或你想更直接地控制：
# 可以先从图中检索相关三元组，然后将这些信息作为上下文喂给 LLM 进行问答。

print("\n--- 使用检索到的上下文进行问答 (备选方案) ---")

def get_relevant_triples(graph, entity_name):
    """一个简单的函数，用于检索与某个实体相关的三元组"""
    relevant_triples_str = []
    for s, p, o in graph.get_triples():
        if entity_name in str(s) or entity_name in str(o):
            relevant_triples_str.append(f"({s}, {p}, {o})")
    return "\n".join(relevant_triples_str)

qa_prompt_template = """根据下面提供的知识图谱信息回答问题。
如果信息不足，请回答“根据已知信息无法回答”。

知识图谱信息：
{context}

问题：{question}
答案："""
qa_prompt = PromptTemplate(template=qa_prompt_template, input_variables=["context", "question"])
qa_chain_with_context = qa_prompt | llm_for_qa

for question in questions:
    print(f"\n问题: {question}")
    # 尝试从问题中提取关键实体 (这本身也可以用LLM完成)
    # 简化的实体提取：
    main_entity = ""
    if "张三" in question: main_entity = "张三"
    elif "李四" in question: main_entity = "李四"
    elif "百度" in question: main_entity = "百度"
    # ... 可以扩展这个逻辑

    if main_entity:
        context_triples = get_relevant_triples(final_graph, main_entity)
        if not context_triples:
            context_triples = "没有找到与该实体直接相关的明确信息。"
    else: # 如果没有明显实体，可以尝试提供所有三元组或一个子集
        context_triples = "\n".join([f"({s}, {p}, {o})" for s, p, o in final_graph.get_triples()[:20]]) # 限制上下文长度

    print(f"提供的上下文: \n{context_triples[:500]}...") # 打印部分上下文
    try:
        answer = qa_chain_with_context.invoke({"context": context_triples, "question": question})
        print(f"答案 (使用上下文): {answer}")
    except Exception as e:
        print(f"回答问题时出错 (使用上下文): {e}")

关于查询的说明:

  • GraphQAChain: 当与支持 Cypher 的图数据库（如 Neo4j）一起使用时，GraphQAChain (特别是 GraphCypherQAChain) 会将自然语言问题转换为 Cypher 查询，在图数据库上执行，然后将结果传递给 LLM 以生成最终答案。对于内存中的 NetworkXEntityGraph，其行为可能更多是基于关键词匹配或简单的图遍历来提取上下文，然后由 LLM 回答。
  • 上下文检索: 对于复杂的查询或 NetworkXEntityGraph，一种更可靠的方法是先设计一个检索步骤：识别问题中的核心实体，从图中检索与这些实体相关的子图或三元组，然后将这些信息作为上下文提供给 LLM 来回答问题。
  • LLM 的角色: 在查询阶段，LLM 不仅用于理解问题，也用于根据检索到的图信息生成流畅、自然的答案。

使用专业的图数据库 (例如 Neo4j)

虽然上面的示例使用了内存中的 NetworkXEntityGraph，但在生产环境中，你通常会使用更强大、持久化的图数据库，如 Neo4j。

使用 Neo4j 的大致流程：

  1. 安装和启动 Neo4j 服务。
  2. 安装 Python 的 Neo4j驱动： pip install neo4j
  3. 连接到 Neo4j：
    from langchain_community.graphs import Neo4jGraph

graph_db = Neo4jGraph(
    url="bolt://localhost:7687", # Neo4j 的 Bolt URL
    username="neo4j",           # 用户名
    password="your_password"    # 密码
)
# 可以清除现有数据（可选，用于测试）
# graph_db.query("MATCH (n) DETACH DELETE n")
  4. 构建图谱：
    • 你可以使用 LLMGraphTransformer 将文档转换为 GraphDocument 对象。
    • 然后使用 graph_db.add_graph_documents(graph_documents) 将这些结构化数据存入 Neo4j。Neo4jGraph 会自动将这些 GraphDocument 转换为 Cypher CREATE 语句并执行。
    • 你也可以直接构造 Cypher 语句来创建节点和关系。
  5. 查询图谱：
    • 使用 GraphCypherQAChain (如果你的 Langchain 版本较旧) 或更新的 GraphQAChain 配置为使用 Cypher。
    • 这个链会将自然语言问题转换为 Cypher 查询。
    from langchain.chains import GraphCypherQAChain # 或者更新的 GraphQAChain

# qa_chain = GraphCypherQAChain.from_llm(
#     cypher_llm=llm_for_qa, # LLM 用于生成 Cypher
#     qa_llm=llm_for_qa,     # LLM 用于从 Cypher 结果生成答案
#     graph=graph_db,
#     verbose=True,
#     # validate_cypher=True, # 验证生成的 Cypher 语句
# )
# 或者使用更新的 API，可能需要指定 GraphDBType
from langchain_community.chains.graph_qa.cypher import GraphCypherQAChain as UpdatedGraphCypherQAChain

# 对于较新版本的 Langchain，可能会有不同的初始化方式或推荐的 Chain
# 检查 Langchain 文档以获取最新的 GraphCypherQAChain 用法

# 假设我们有一个可以生成 Cypher 的 LLM 和一个可以从结果回答问题的 LLM
# (这里简化，两者都用 llm_for_qa)
cypher_generation_llm = llm_for_qa
qa_llm = llm_for_qa

chain = UpdatedGraphCypherQAChain.from_llm(
    graph=graph_db,
    cypher_llm=cypher_generation_llm,
    qa_llm=qa_llm,
    validate_cypher=True, # 推荐在开发时开启
    verbose=True
)

# question = "张三研究什么领域？"
# result = chain.invoke({"query": question})
# print(result["result"])
    • Cypher 生成的挑战： 让 LLM 准确生成符合你的图谱 Schema 的 Cypher 查询可能具有挑战性，需要良好的提示工程和图谱 Schema 的描述。GraphCypherQAChain 内部包含提示来指导 LLM 根据图的 Schema 生成 Cypher。你需要确保 graph_db.schema (或通过 refresh_schema=True 更新的 schema) 对 LLM 可见且准确。

总结与进一步探索

  • 选择合适的图存储： 对于小型实验，NetworkXEntityGraph 很方便。对于生产应用，应选择专业的图数据库（Neo4j, NebulaGraph, Amazon Neptune, Kùzu 等）。
  • Prompt Engineering is Key： 无论是实体/关系提取还是 Cypher 生成，提示的质量都直接影响结果。针对阿里云百炼模型，你可能需要进行特定的提示词优化。
  • Schema 设计： 一个良好定义的图谱 Schema (节点标签、属性、关系类型) 对于后续的查询和分析至关重要。
  • 评估与迭代： 构建和查询知识图谱是一个迭代的过程。你需要持续评估提取的准确性、图谱的完整性以及查询的性能和结果质量。
  • 高级特性：
    • Graph RAG (Retrieval Augmented Generation): 将知识图谱作为上下文信息源，增强 LLM 的问答能力。KnowledgeGraphIndex 可以用于此目的。
    • 复杂推理： 在图上执行更复杂的推理任务，例如路径查找、社区检测、链接预测等。
    • 多跳查询 (Multi-hop Queries): 回答需要连接多个信息片段才能得到的复杂问题。
LangChain
最后修改于 May 20, 2025 15:48 +0800
© 2024 - 2025 Example Person
使用 Hugo 构建
主题 Stack 由 Jimmy 设计