RAG正在进化，但真正的突破不在参数，而在时间维度。如何让AI代理理解“知识的时间性”？如何在系统中嵌入“时态感知”的机制？本文以产品视角切入，提出“动作-时间-知识”三元协同模型，为AI产品经理提供下一代知识系统的设计思路。

用于回答问题的 RAG 或代理架构依赖于随着时间的推移不断更新的动态知识库，例如财务报告或文档，以便推理和规划步骤保持逻辑和准确。

为了处理这样的知识库，其中规模不断增长，幻觉的机会可能会增加，需要一个单独的逻辑时间（时间感知）代理管道来管理 AI 产品中这个不断发展的知识库 。

1. 百分位语义分块：将大型原始文档分解为具有上下文意义的小文本块；
2. 原子事实：使用LLM读取每个块并提取原子事实、它们的时间戳和所涉及的实体；
3. 实体解析：通过自动查找和合并重复实体（例如，“AMD”和“AdvancedMicroDevices”）来清理数据；
4. 时间无效：当新信息到达时，通过将过时的事实标记为“过期”来智能识别和解决矛盾；
5. 知识图谱构建：将最终的、干净的、带有时间戳的事实组装成一个连接的图形结构，我们的AI代理可以查询；
6. 优化的知识库：将最终的动态知识图谱存储在可扩展的云数据库中，创建可靠、最新的“大脑”，在此基础上构建最终的RAG或代理系统；

预处理和分析我们的动态数据

公司定期分享其财务业绩的最新信息，例如股价走势、执行领导层变动等重大发展，以及前瞻性预期，例如季度收入是否预计同比增长 12% 等。

在医疗领域，ICD 编码是数据不断发展的另一个例子，从 ICD-9 到 ICD-10 的过渡使诊断代码从约 14,000 个增加到 68,000 个。

让我们加载这个数据集并对其进行一些统计分析以适应它。

# Import loader for Hugging Face datasets

from langchain_community.document_loaders import HuggingFaceDatasetLoader

# Dataset configuration

hf_dataset_name = “jlh-ibm/earnings_call” # HF dataset name

subset_name = “transcripts” # Dataset subset to load

# Create the loader (defaults to ‘train’ split)

loader = HuggingFaceDatasetLoader(

path=hf_dataset_name,

name=subset_name,

page_content_column=”transcript” # Column containing the main text

)

# This is the key step. The loader processes the dataset and returns a list of LangChain Document objects.

documents = loader.load()

我们专注于该数据集的成绩单子集，其中包含有关不同公司的原始文本信息。这是数据集的基本结构，可作为任何 RAG 或 AI 代理架构的起点。

# Let’s inspect the result to see the difference

print(f”Loaded {len(documents)} documents.”)

#

#

## OUTPUT ####

Loaded 188 documents.

我们的数据中共有 188 份成绩单。这些成绩单属于不同的公司，我们需要计算我们的数据集中代表了多少个独特的公司。

# Count how many documents each company has

company_counts = {}

# Loop over all loaded documents

for doc in documents:

company = doc.metadata.get(“company”) # Extract company from metadata

if company:

company_counts[company] = company_counts.get(company, 0) + 1

# Display the counts

print(“Total company counts:”)

for company, count in company_counts.items():

print(f”

– {company}: {count}”)

#

#

## OUTPUT ####

Total company counts:

– AMD: 19

– AAPL: 19

– INTC: 19

– MU: 17

– GOOGL: 19

– ASML: 19

– CSCO: 19

– NVDA: 19

– AMZN: 19

– MSFT: 19

几乎所有公司的分配比例都相等。查看随机成绩单的元数据。

# Print metadata for two sample documents (index 0 and 33)

print(“Metadata for document[0]:”)

print(documents[0].metadata)

print(“\nMetadata for document[33]:”)

print(documents[33].metadata)

#

#

## OUTPUT ####

{‘company’: ‘AMD’, ‘date’: datetime.date(2016, 7, 21)}

{‘company’: ‘AMZN’, ‘date’: datetime.date(2019, 10, 24)}

公司字段仅指示成绩单属于哪家公司， 日期字段表示信息所基于的时间范围。

# Print the first 200 characters of the first document’s content

first_doc = documents[0]

print(first_doc.page_content[:200])

#

#

## OUTPUT ####

Thomson Reuters StreetEvents Event Transcript

E D I T E D V E R S I O N

Q2 2016 Advanced Micro Devices Inc Earnings Call

JULY 21, 2016 / 9:00PM GMT

=====================================

通过打印文档的样本，我们可以获得高级概述。例如，当前示例显示了 AMD 的季度报告。

成绩单可能很长，因为它们代表给定时间范围内的信息并包含大量详细信息。我们需要检查这 188 份成绩单平均包含多少个单词。

# Calculate the average number of words per document

total_words = sum(len(doc.page_content.split()) for doc in documents)

average_words = total_words / len(documents) if documents else 0

print(f”Average number of words in documents: {average_words:.2f}”)

#

#

## OUTPUT ####

Average number of words in documents: 8797.124

每个成绩单 ~9K 字是相当大的，因为它肯定包含大量信息。但这正是我们所需要的，创建一个结构良好的知识库 AI 代理涉及处理大量信息，而不仅仅是几个小文档。

通常，财务数据基于不同的时间范围，每个时间范围代表有关该时期发生情况的不同信息。我们可以使用纯 Python 代码而不是 LLM 从成绩单中提取这些时间范围，以节省成本。

import re

from datetime import datetime

# Helper function to extract a quarter string (e.g., “Q1 2023”) from text

def find_quarter(text: str) -> str | None:

“””Return the first quarter-year match found in the text, or None if absent.”””

# Match pattern: ‘Q’ followed by 1 digit, a space, and a 4-digit year

match = re.findall(r”Q\d\s\d{4}”, text)

return match[0] if match else None

# Test on the first document

quarter = find_quarter(documents[0].page_content)

print(f”Extracted Quarter for the first document: {quarter}”)

#

#

## OUTPUT ####

Extracted Quarter for the first document: Q2 2016

执行季度日期提取的更好方法是通过法学硕士，因为他们可以更深入地理解数据。但是，由于我们的数据在文本方面已经结构良好，因此我们暂时可以在没有它们的情况下继续进行。

百分位语义分块

通常，我们根据随机拆分或有意义的句子边界（例如以句号结尾）对数据进行分块。但是，这种方法可能会导致丢失一些信息。

如果我们在句号处拆分，我们就失去了净收入增长是由于运营费用下降的紧凑联系。

我们将在这里使用基于百分位数的分块。让我们先了解这种方法，然后再实施它。

• 该文档使用正则表达式拆分为句子，通常在.、？或！之后中断。
• 每个句子都使用嵌入模型转换为高维向量。
• 计算连续句子向量之间的语义距离，值越大表示主题变化越大。
• 收集所有距离，并确定所选的百分位数（例如95个百分位数）以捕获异常大的跳跃。
• 距离大于或等于此阈值的边界标记为块断点。
• 这些边界之间的句子被分组为块，应用min_chunk_size以避免过小的块，并在需要时buffer_size添加重叠。

from langchain_nebius import NebiusEmbeddings

# Set Nebius API key (⚠️ Avoid hardcoding secrets in production code)

os.environ[“NEBIUS_API_KEY”] = “YOUR_API_KEY_HERE”

#

1. Initialize Nebius embedding model

embeddings = NebiusEmbeddings(model=”Qwen/Qwen3-Embedding-8B”)

我们正在使用 Qwen3–8B 通过 LangChain 中的 Nebius AI 生成嵌入。当然，LangChain 模块下还支持许多其他嵌入提供者。

from langchain_experimental.text_splitter import SemanticChunker

# Create a semantic chunker using percentile thresholding

langchain_semantic_chunker = SemanticChunker(

embeddings,

breakpoint_threshold_type=”percentile”, # Use percentile-based splitting

breakpoint_threshold_amount=95 # split at 95th percentile

)

我们选择了第 95 个百分位值，这意味着如果连续句子之间的距离超过该值，则将其视为断点。使用循环，我们可以简单地在成绩单上启动分块过程。

# Store the new, smaller chunk documents

chunked_documents_lc = []

# Printing total number of docs (188) We already know that

print(f”Processing {len(documents)} documents using LangChain’s SemanticChunker…”)

# Chunk each transcript document

for doc in tqdm(documents, desc=”Chunking Transcripts with LangChain”):

# Extract quarter info and copy existing metadata

quarter = find_quarter(doc.page_content)

parent_metadata = doc.metadata.copy()

parent_metadata[“quarter”] = quarter

# Perform semantic chunking (returns Document objects with metadata attached)

chunks = langchain_semantic_chunker.create_documents(

[doc.page_content],

metadatas=[parent_metadata]

)

# Collect all chunks

chunked_documents_lc.extend(chunks)

#

#

## OUTPUT ####

Processing 188 documents using LangChains SemanticChunker…

Chunking Transcripts with LangChain: 100%|██████████| 188/188 [01:03:44<00:00, 224.91s/it]

这个过程将花费很多时间，因为正如我们之前看到的，每个成绩单大约有 8K 字。为了加快速度，我们可以使用异步函数来减少运行时间。但是，为了更容易理解，这个循环完全符合我们想要实现的目的。

# Analyze the results of the LangChain chunking process

original_doc_count = len(docs_to_process)

chunked_doc_count = len(chunked_documents_lc)

print(f”Original number of documents (transcripts): {original_doc_count}”)

print(f”Number of new documents (chunks): {chunked_doc_count}”)

print(f”Average chunks per transcript: {chunked_doc_count / original_doc_count:.2f}”)

#

#

## OUTPUT ####

Original number of documents (transcripts): 188

Number of new documents (chunks): 3556

Average chunks per transcript: 19.00

平均而言，我们每个成绩单有 19 个块。让我们检查一下我们的一个成绩单中的随机块。

# Inspect the 11th chunk (index 10)

sample_chunk = chunked_documents_lc[10]

print(“Sample Chunk Content (first 30 chars):”)

print(sample_chunk.page_content[:30] + “…”)

print(“\nSample Chunk Metadata:”)

print(sample_chunk.metadata)

# Calculate average word count per chunk

total_chunk_words = sum(len(doc.page_content.split()) for doc in chunked_documents_lc)

average_chunk_words = total_chunk_words / chunked_doc_count if chunked_documents_lc else 0

print(f”\nAverage number of words per chunk: {average_chunk_words:.2f}”)

#

#

## OUTPUT ####

Sample Chunk Content (first 30 chars):

No, that is a fair question, Matt. So we have been very focused …

Sample Chunk Metadata:

{‘company’: ‘AMD’, ‘date’: datetime.date(2016, 7, 21), ‘quarter’: ‘Q2 2016’}

Average number of words per chunk: 445.42

我们的块数据元数据略有变化，包括一些附加信息，例如块所属的季度，以及 Python 友好格式的日期时间，以便于检索。

使用语句代理提取原子事实

现在我们已经将数据整齐地组织成有意义的小块，我们可以开始使用 LLM 来读取这些块并提取核心事实。

我们需要将文本分解为尽可能小的“原子”事实。例如，我们想要的不是单个复杂的句子，而是可以独立存在的个别主张。

这个过程使我们的人工智能系统以后更容易理解、查询和推理信息。

为了确保我们的 LLM 为我们提供干净、可预测的输出，我们需要给它一套严格的指令。在 Python 中执行此作的最佳方法是使用 Pydantic 模型。这些模型充当 LLM 必须遵循的“模式”或“模板”。

首先，让我们使用枚举定义标签允许的类别。我们正在使用 （str， Enum）。

from enum import Enum

# Enum for temporal labels describing time sensitivity

class TemporalType(str, Enum):

ATEMPORAL = “ATEMPORAL” # Facts that are always true (e.g., “Earth is a planet”)

STATIC = “STATIC” # Facts about a single point in time (e.g., “Product X launched on Jan 1st”)

DYNAMIC = “DYNAMIC” # Facts describing an ongoing state (e.g., “Lisa Su is the CEO”)

每个类别都捕获不同类型的时间参考：

• 非时间：普遍正确且随时间不变的陈述（例如，“水在100摄氏度时沸腾”）；
• 静态的：在特定时间点成为事实并此后保持不变的陈述（例如，“John于2020年6月1日被聘为经理”）；
• 动态：可能会随着时间的推移而变化并需要时间上下文才能准确解释的陈述（例如，“约翰是团队的经理”）；

# Enum for statement labels classifying statement nature

class StatementType(str, Enum):

FACT = “FACT” # An objective, verifiable claim

OPINION = “OPINION” # A subjective belief or judgment

PREDICTION = “PREDICTION” # A statement about a future event

StatementType 枚举显示每个语句的类型。

1. 事实：当时的陈述是真实的，但以后可能会改变（例如，“公司上个季度赚了500万美元”）；
2. 意见：个人信念或感受，只有在说出来时才是真实的（例如，“我认为这个产品会做得很好”）；
3. 预测：对未来的猜测，从现在到预测时间结束都是正确的（例如，“明年的销售额将增长”）；

通过定义这些固定类别，我们确保代理对其提取的信息进行分类的方式的一致性。

现在，让我们创建 Pydantic 模型，该模型将使用这些枚举来定义输出的结构。

from pydantic import BaseModel, field_validator

# This model defines the structure for a single extracted statement

class RawStatement(BaseModel):

statement: str

statement_type: StatementType

temporal_type: TemporalType

# This model is a container for the list of statements from one chunk

class RawStatementList(BaseModel):

statements: list[RawStatement]

这些模型是我们与 LLM 的合同。我们告诉它，“当你处理完一个块时，你的答案必须是一个 JSON 对象，其中包含一个名为”语句“的列表，并且该列表中的每个项目都必须有一个语句 、一个 statement_type 和一个 temporal_type”。

这些定义只是对每个标签的描述。他们通过解释为什么 LLM 应该为提取的信息分配特定标签，为 LLM 提供更好的支持。现在，我们需要使用这些定义创建提示模板。

# Format label definitions into a clean string for prompt injection

definitions_text = “”

for section_key, section_dict in LABEL_DEFINITIONS.items():

# Add a section header with underscores replaced by spaces and uppercased

definitions_text += f”==== {section_key.replace(‘_’, ‘ ‘).upper()} DEFINITIONS ====\n”

# Add each category and its definition under the section

for category, details in section_dict.items():

definitions_text += f”

– {category}: {details.get(‘definition’, ”)}\n”

这个 definitions_text 字符串将是我们提示的关键部分，为 LLM 提供准确执行任务所需的“教科书”定义。

现在，我们将构建主提示模板。该模板将所有内容汇集在一起：输入、任务指令、标签定义以及一个关键示例，以向 LLM 展示良好输出的确切外观。

请注意我们如何使用 {{ 和 }} 在 JSON 示例中使用大括号“转义”。这告诉 LangChain 这些是文本的一部分，而不是要填充的变量。

from langchain_core.prompts import ChatPromptTemplate

# Define the prompt template for statement extraction and labeling

statement_extraction_prompt_template = “””

You are an expert extracting atomic statements from text.

Inputs:

– main_entity: {main_entity}

– document_chunk: {document_chunk}

Tasks:

1. Extract clear, single-subject statements.

2. Label each as FACT, OPINION, or PREDICTION.

3. Label each temporally as STATIC, DYNAMIC, or ATEMPORAL.

4. Resolve references to main_entity and include dates/quantities.

Return ONLY a JSON object with the statements and labels.

“””

# Create a ChatPromptTemplate from the string template

prompt = ChatPromptTemplate.from_template(statement_extraction_prompt_template)

最后，我们将一切连接起来。我们将创建一个 LangChain“链”，将我们的提示链接到 LLM，并告诉 LLM 根据我们的 RawStatementList 模型构建其输出。

我们将通过 Nebius 使用 deepseek-ai/DeepSeek-V3 模型来完成此任务，因为它功能强大且擅长遵循复杂的指令。

from langchain_nebius import ChatNebius

import json

# Initialize our LLM

llm = ChatNebius(model=”deepseek-ai/DeepSeek-V3″)

# Create the chain: prompt -> LLM -> structured output parser

statement_extraction_chain = prompt | llm.with_structured_output(RawStatementList)

到目前为止，我们已经完成了原子事实步骤，我们准确地提取了时间和基于语句的事实，这些事实可以在以后根据需要进行更新。

使用验证检查代理精确定位时间

我们已经成功地从文本中提取了原子语句。现在，我们需要提取 when。每个语句都需要一个精确的时间戳来告诉我们它何时有效。

产品是上周推出还是去年推出？或者 CEO 是在 2016 年或 2018 年担任过她的职务吗？

这是使我们的知识库真正“时间化”的最重要一步。

从自然语言中提取日期很棘手。一份声明可能会说下个季度 、 三个月前或 2017 年 。人类理解这一点，但计算机需要一个具体的日期，比如 2017-01-01。

我们的目标是创建一个专门的代理，它可以读取声明，并使用原始文档的发布日期作为参考，找出两个关键时间戳：

• valid_at：事实成为现实的日期；
• invalid_at：事实不再为真的日期（如果它不再有效）；

就像以前一样，我们将首先定义 Pydantic 模型，以确保我们的 LLM 为我们提供干净、结构化的日期输出。

首先，我们需要一个强大的辅助函数来解析 LLM 可能返回的各种日期格式。此函数可以处理 2017 或 2016–07–21 等字符串，并将它们转换为正确的 Python 日期时间对象。

from pydantic import BaseModel, Field, field_validator

from datetime import datetime

# Model for raw temporal range with date strings as ISO 8601

class RawTemporalRange(BaseModel):

valid_at: str | None = Field(None, description=”The start date/time in ISO 8601 format.”)

invalid_at: str | None = Field(None, description=”The end date/time in ISO 8601 format.”)

# Model for validated temporal range with datetime objects

class TemporalValidityRange(BaseModel):

valid_at: datetime | None = None

invalid_at: datetime | None = None

# Validator to parse date strings into datetime objects before assignment

@field_validator(“valid_at”, “invalid_at”, mode=”before”)

@classmethod

def _parse_date_string(cls, value: str | datetime | None) -> datetime | None:

return parse_date_str(value)

这种两步模型方法是一个很好的实践。它将原始 LLM 输出与我们干净、经过验证的应用程序数据分开，使管道更加健壮。

接下来，我们专门为此日期提取任务创建一个新提示。我们将向 LLM 提供我们提取的语句之一，并要求它根据上下文确定有效日期。

# Prompt guiding the LLM to extract temporal validity ranges from statements

date_extraction_prompt_template = “””

You are a temporal information extraction specialist.

INPUTS:

– statement: “{statement}”

– statement_type: “{statement_type}”

– temporal_type: “{temporal_type}”

– publication_date: “{publication_date}”

– quarter: “{quarter}”

TASK:

– Analyze the statement and determine the temporal validity range (valid_at, invalid_at).

– Use the publication date as the reference point for relative expressions (e.g., “currently”).

– If a relationship is ongoing or its end is not specified, `invalid_at` should be null.

GUIDANCE:

– For STATIC statements, `valid_at` is the date the event occurred, and `invalid_at` is null.

– For DYNAMIC statements, `valid_at` is when the state began, and `invalid_at` is when it ended.

– Return dates in ISO 8601 format (e.g., YYYY-MM-DDTHH:MM:SSZ).

**Output format**

Return ONLY a valid JSON object matching the schema for `RawTemporalRange`.

“””

# Create a LangChain prompt template from the string

date_extraction_prompt = ChatPromptTemplate.from_template(date_extraction_prompt_template)

此提示非常专业化。它告诉 LLM 像“时间专家”一样行事，并为如何处理不同类型的语句提供了明确的规则。

现在我们为此步骤构建 LangChain 链，并在我们之前提取的语句之一上对其进行测试。

# Reuse the existing LLM instance.

# Create a chain by connecting the date extraction prompt

# with the LLM configured to output structured RawTemporalRange objects.

date_extraction_chain = date_extraction_prompt | llm.with_structured_output(RawTemporalRange)

这是一个动态事实，因此我们预计日期 valid_at，但未指定结束日期。

LLM 正确解释了“2017 年上半年”，并将其转换为精确的日期范围。它明白，这个动态声明有明确的开始和结束。

我们现在已经成功地在我们的事实中添加了时间维度。下一步是将陈述的文本进一步分解为知识图谱的基本结构，即三元组 。

将事实结构化为三元组

现在，我们需要将这些自然语言句子转换为人工智能代理可以轻松理解和连接的格式。

三元组将一个事实分解为三个核心组成部分：

1. 主题：事实是关于的主要实体；
2. 谓语：关系或行动；
3. 对象：主题相关的实体或概念；

通过将我们所有的陈述转换为这种格式，我们可以构建一个相互关联的事实网络，即我们的知识图谱。

和以前一样，我们将首先定义 Pydantic 模型，这些模型将构建 LLM 的输出。这种提取是我们迄今为止最复杂的，因为 LLM 需要同时识别实体（名词）和关系（三元组）。

首先，让我们定义我们希望代理使用的谓词关系的固定列表。这确保了我们整个知识图谱的一致性。

from enum import Enum # Import the Enum base class to create enumerated constants

# Enum representing a fixed set of relationship predicates for graph consistency

class Predicate(str, Enum):

# Each member of this Enum represents a specific type of relationship between entities

IS_A = “IS_A” # Represents an “is a” relationship (e.g., a Dog IS_A Animal)

HAS_A = “HAS_A” # Represents possession or composition (e.g., a Car HAS_A Engine)

LOCATED_IN = “LOCATED_IN” # Represents location relationship (e.g., Store LOCATED_IN City)

HOLDS_ROLE = “HOLDS_ROLE” # Represents role or position held (e.g., Person HOLDS_ROLE Manager)

PRODUCES = “PRODUCES” # Represents production or creation relationship

SELLS = “SELLS” # Represents selling relationship between entities

LAUNCHED = “LAUNCHED” # Represents launch events (e.g., Product LAUNCHED by Company)

DEVELOPED = “DEVELOPED” # Represents development relationship (e.g., Software DEVELOPED by Team)

ADOPTED_BY = “ADOPTED_BY” # Represents adoption relationship (e.g., Policy ADOPTED_BY Organization)

INVESTS_IN = “INVESTS_IN” # Represents investment relationships (e.g., Company INVESTS_IN Startup)

COLLABORATES_WITH = “COLLABORATES_WITH” # Represents collaboration between entities

SUPPLIES = “SUPPLIES” # Represents supplier relationship (e.g., Supplier SUPPLIES Parts)

HAS_REVENUE = “HAS_REVENUE” # Represents revenue relationship for entities

INCREASED = “INCREASED” # Represents an increase event or metric change

DECREASED = “DECREASED” # Represents a decrease event or metric change

RESULTED_IN = “RESULTED_IN” # Represents causal relationship (e.g., Event RESULTED_IN Outcome)

TARGETS = “TARGETS” # Represents target or goal relationship

PART_OF = “PART_OF” # Represents part-whole relationship (e.g., Wheel PART_OF Car)

DISCONTINUED = “DISCONTINUED” # Represents discontinued status or event

SECURED = “SECURED” # Represents secured or obtained relationship (e.g., Funding SECURED by Company)

现在，我们定义了原始实体和三元组的模型，以及一个容器模型 RawExtraction，以保存整个输出。

from pydantic import BaseModel, Field

from typing import List, Optional

# Model representing an entity extracted by the LLM

class RawEntity(BaseModel):

entity_idx: int = Field(description=”A temporary, 0-indexed ID for this entity.”)

name: str = Field(description=”The name of the entity, e.g., ‘AMD’ or ‘Lisa Su’.”)

type: str = Field(“Unknown”, description=”The type of entity, e.g., ‘Organization’, ‘Person’.”)

description: str = Field(“”, description=”A brief description of the entity.”)

# Model representing a single subject-predicate-object triplet

class RawTriplet(BaseModel):

subject_name: str

subject_id: int = Field(description=”The entity_idx of the subject.”)

predicate: Predicate

object_name: str

object_id: int = Field(description=”The entity_idx of the object.”)

value: Optional[str] = Field(None, description=”An optional value, e.g., ‘10%’.”)

# Container for all entities and triplets extracted from a single statement

class RawExtraction(BaseModel):

entities: List[RawEntity]

triplets: List[RawTriplet]

这个结构非常巧妙。它要求 LLM 首先列出它找到的所有实体，并为每个实体提供一个临时编号 （entity_idx）。

然后，它要求法学硕士使用这些数字构建三元组，在关系和所涉及的实体之间建立清晰的联系。

接下来，我们创建将指导 LLM 的提示和定义。此提示非常具体，指示模型仅关注关系，忽略我们已经提取的任何基于时间的信息。

现在是主要提示模板。同样，我们小心翼翼地使用 {{ 和 }} 转义 JSON 示例，以便 LangChain 可以正确解析它。

# Prompt for extracting entities and subject-predicate-object triplets from a statement

triplet_extraction_prompt_template = “””

You are an information-extraction assistant.

Task: From the statement, identify all entities (people, organizations, products, concepts) and all triplets (subject, predicate, object) describing their relationships.

Statement: “{statement}”

Predicate list:

{predicate_instructions}

Guidelines:

– List entities with unique `entity_idx`.

– List triplets linking subjects and objects by `entity_idx`.

– Exclude temporal expressions from entities and triplets.

Example:

Statement: “Google’s revenue increased by 10% from January through March.”

Output: {{

“entities”: [

{{“entity_idx”: 0, “name”: “Google”, “type”: “Organization”, “description”: “A multinational technology company.”}},

{{“entity_idx”: 1, “name”: “Revenue”, “type”: “Financial Metric”, “description”: “Income from normal business.”}}

],

“triplets”: [

{{“subject_name”: “Google”, “subject_id”: 0, “predicate”: “INCREASED”, “object_name”: “Revenue”, “object_id”: 1, “value”: “10%”}}

]

}}

Return ONLY a valid JSON object matching `RawExtraction`.

“””

# Initializing the prompt template

triplet_extraction_prompt = ChatPromptTemplate.from_template(triplet_extraction_prompt_template)

最后，我们创建第三条链并在我们的一个语句上对其进行测试。

# Create the chain for triplet and entity extraction.

triplet_extraction_chain = triplet_extraction_prompt | llm.with_structured_output(RawExtraction)

# Let’s use the same statement we’ve been working with.

sample_statement_for_triplets = extracted_statements_list.statements[0]

print(f”–

– Running triplet extraction for statement —“)

print(f’Statement: “{sample_statement_for_triplets.statement}”‘)

# Invoke the chain.

raw_extraction_result = triplet_extraction_chain.invoke({

“statement”: sample_statement_for_triplets.statement,

“predicate_instructions”: predicate_instructions_text

})

print(“\n–

– Triplet Extraction Result —“)

print(raw_extraction_result.model_dump_json(indent=2))

LLM 正确地将“AMD”和“server launch”识别为关键实体，并将它们与 TARGETS 谓词联系起来，完美地捕捉到了原句的含义。

我们现在已经完成了所有单独的提取步骤。我们有一个系统，可以获取一段文本并提取语句、日期、实体和三元组。下一步是将所有这些信息组合成一个统一的对象，该对象表示一个完整的 “时间事件”。

组装时态事件

我们现在已经完成了所有单独的提取步骤。我们有一个系统，可以获取一段文本并提取：

1. 陈述：原子事实；
2. 日期：每个事实的“时间”；
3. 实体和三元组：结构化格式的“谁”和“什么”；

我们提取过程的最后一步是将所有这些部分放在一起。我们将创建一个名为 TemporalEvent 的主数据模型，该模型将有关单个语句的所有信息合并到一个干净、统一的对象中。

此 TemporalEvent 将是我们在引入管道的其余部分中使用的中心对象。它将包含所有内容：原始语句、其类型、其时间范围以及指向从中派生的所有三元组的链接。

首先，我们将 RawEntity 和 RawTriplet 对象转换为最终的持久实体和三元组形式，并配有唯一的 UUID。

# Assume these are already defined from previous steps:

# sample_statement, final_temporal_range, raw_extraction_result

print(“–

– Assembling the final TemporalEvent —“)

#

1. Convert raw entities to persistent Entity objects with UUIDs

idx_to_entity_map: dict[int, Entity] = {}

final_entities: list[Entity] = []

for raw_entity in raw_extraction_result.entities:

entity = Entity(

name=raw_entity.name,

type=raw_entity.type,

description=raw_entity.description

)

idx_to_entity_map[raw_entity.entity_idx] = entity

final_entities.append(entity)

print(f”Created {len(final_entities)} persistent Entity objects.”)

#

2. Convert raw triplets to persistent Triplet objects, linking entities via UUIDs

final_triplets: list[Triplet] = []

for raw_triplet in raw_extraction_result.triplets:

subject_entity = idx_to_entity_map[raw_triplet.subject_id]

object_entity = idx_to_entity_map[raw_triplet.object_id]

triplet = Triplet(

subject_name=subject_entity.name,

subject_id=subject_entity.id,

predicate=raw_triplet.predicate,

object_name=object_entity.name,

object_id=object_entity.id,

value=raw_triplet.value

)

final_triplets.append(triplet)

print(f”Created {len(final_triplets)} persistent Triplet objects.”)

我们已经成功地跟踪了一条信息，从原始段落一直到完全结构化的、带有时间戳的 TemporalEvent。

但是对每个语句手动执行此作是不可能的。下一步是使用 LangGraph 自动化整个装配线，以一次处理我们所有的块。

使用 LangGraph 自动化管道

到目前为止，我们已经构建了三个强大的、专门的“代理”（或链）：一个用于提取语句，一个用于日期，一个用于三元组。我们还了解了如何手动将它们的输出组合成最终的 TemporalEvent。

构建 LangGraph 的第一步是定义其“状态”。状态是图的内存，它是在每个节点之间传递的数据。我们将定义一个状态，该状态可以保存我们的块列表和我们在此过程中创建的所有新对象。

from typing import List, TypedDict

from langchain_core.documents import Document

class GraphState(TypedDict):

“””

TypedDict representing the overall state of the knowledge graph ingestion.

Attributes:

chunks: List of Document chunks being processed.

temporal_events: List of TemporalEvent objects extracted from chunks.

entities: List of Entity objects in the graph.

triplets: List of Triplet objects representing relationships.

“””

chunks: List[Document]

temporal_events: List[TemporalEvent]

entities: List[Entity]

triplets: List[Triplet]

现在，我们将把之前的所有逻辑组合成一个强大的函数。该函数将是我们图表中的主要“节点”。它从状态中获取块列表，并通过三个并行步骤编排整个提取过程。

我们已经成功地实现了提取管道的自动化。然而，数据仍然是“原始的”。例如，LLM 可能已将“AMD”和“Advanced Micro Devices”提取为两个独立的实体。下一步是在我们的装配线上添加一个质量控制站： 实体解析 。

使用实体解析清理我们的数据

我们的自动化管道现在正在提取大量信息。但是，如果您仔细观察实体，您可能会注意到一个问题。

这是一个关键问题。如果我们不修复它，我们的知识图谱就会变得混乱和不可靠。有关“AMD”的查询会遗漏与“Advanced Micro Devices”相关的事实。

这称为实体解析 。目标是识别引用同一现实世界实体的所有不同名称（或“提及”），并将它们合并到一个单一的、权威的、“规范”的 ID 下。

为了解决这个问题，我们将在我们的 LangGraph 装配线上添加一个新的质量控制站。该节点将：

1. 使用模糊字符串匹配对具有相似名称的实体进行集群；
2. 为群集中的所有实体分配一个规范ID；
3. 更新我们的三胞胎以使用这些新的、干净的ID；

为了跟踪我们的规范实体，我们需要一个地方来存储它们。在本教程中，我们将使用 Python 的内置 sqlite3 库创建一个简单的内存数据库。在实际应用程序中，这将是一个持久的生产级数据库。

让我们创建内存数据库和我们需要的表。

import sqlite3

def setup_in_memory_db():

“””

Sets up an in-memory SQLite database and creates the ‘entities’ table.

The ‘entities’ table schema:

– id: TEXT, Primary Key

– name: TEXT, name of the entity

– type: TEXT, type/category of the entity

– description: TEXT, description of the entity

– is_canonical: INTEGER, flag to indicate if entity is canonical (default 1)

Returns:

sqlite3.Connection: A connection object to the in-memory database.

“””

# Establish connection to an in-memory SQLite database

conn = sqlite3.connect(“:memory:”)

# Create a cursor object to execute SQL commands

cursor = conn.cursor()

# Create the ‘entities’ table if it doesn’t already exist

cursor.execute(“””

CREATE TABLE IF NOT EXISTS entities (

id TEXT PRIMARY KEY,

name TEXT,

type TEXT,

description TEXT,

is_canonical INTEGER DEFAULT 1

)

“””)

# Commit changes to save the table schema

conn.commit()

# Return the connection object for further use

return conn

# Create the database connection and set up the entities table

db_conn = setup_in_memory_db()

这个简单的数据库将充当我们的代理识别的所有干净、权威实体的中央注册表。

现在我们将为新的 LangGraph 节点创建函数。此函数将包含用于查找和合并重复实体的逻辑。对于模糊字符串匹配，我们将使用一个名为 rapidfuzz 的便捷库。您可以简单地使用 pip install rapidfuzz 安装它。

这是解析的实体输出：

# —

– Sample of a Resolved Entity —

{

“id”: “1a2b3c4d-5e6f-4a7b-8c9d-0e1f2a3b4c5d”,

“name”: “Advanced Micro Devices”,

“type”: “Organization”,

“description”: “A semiconductor company.”,

“resolved_id”: “b1c2d3e4-f5g6-4h7i-8j9k-0l1m2n3o4p5q”

}

# —

– Sample Triplet with Resolved IDs —

{

“id”: “c1d2e3f4-a5b6-4c7d-8e9f-0g1h2i3j4k5l”,

“subject_name”: “AMD”,

“subject_id”: “b1c2d3e4-f5g6-4h7i-8j9k-0l1m2n3o4p5q”,

“predicate”: “TARGETS”,

“object_name”: “server launch”,

“object_id”: “d1e2f3a4-b5c6-4d7e-8f9g-0h1i2j3k4l5m”,

“value”: null

}

这太完美了。您可以看到，名为“Advanced Micro Devices”的实体现在有一个指向规范实体 ID 的 resolved_id（可能是名为“AMD”的实体）。我们所有的三胞胎现在都使用这些干净的规范 ID。

我们的数据现在不仅结构化且带有时间戳，而且干净且一致。下一步是我们代理最聪明的部分：处理与失效代理的矛盾。

使用无效代理使我们的知识充满活力

我们的数据现在被分块、提取、结构化和清理。但我们仍然没有解决时间数据的核心挑战：事实会随着时间的推移而变化。

想象一下，我们的知识库包含以下事实： (John Smith) –[HOLDS_ROLE]–> (CFO) .这是一个动态的事实，这意味着它在一段时间内是正确的。当我们的代理人阅读一份新文件时，会发生什么，上面写着 “Jane Doe 于 2024 年 1 月 1 日被任命为首席财务官”？

第一个事实现在已经过时了。一个简单的知识库不会知道这一点，但一个临时的知识库必须知道这一点。这就是无效代理的工作：充当裁判，发现这些矛盾，并更新旧事实以将其标记为过期。

为此，我们将向 LangGraph 管道添加一个最终节点。该节点将：

1. 为我们所有的新语句生成嵌入，以理解它们的语义含义；
2. 将新的动态事实与我们数据库中的现有事实进行比较；
3. 使用LLM对新事实是否使旧事实无效做出最终判断；
4. 如果事实无效，它将更新其invalid_at时间戳；

首先，我们需要为失效逻辑准备我们的环境。这涉及将事件表和三元组表添加到我们的内存数据库中。这模拟了代理可以检查的真实、持久的知识库。

# Obtain a cursor from the existing database connection

cursor = db_conn.cursor()

# Create the ‘events’ table to store event-related data

cursor.execute(“””

CREATE TABLE IF NOT EXISTS events (

id TEXT PRIMARY KEY, –

– Unique identifier for each event

chunk_id TEXT, –

– Identifier for the chunk this event belongs to

statement TEXT, –

– Textual representation of the event

statement_type TEXT, –

– Type/category of the statement (e.g., assertion, question)

temporal_type TEXT, –

– Temporal classification (e.g., past, present, future)

valid_at TEXT, –

– Timestamp when the event becomes valid

invalid_at TEXT, –

– Timestamp when the event becomes invalid

embedding BLOB –

– Optional embedding data stored as binary (e.g., vector)

)

“””)

# Create the ‘triplets’ table to store relations between entities for events

cursor.execute(“””

CREATE TABLE IF NOT EXISTS triplets (

id TEXT PRIMARY KEY, –

– Unique identifier for each triplet

event_id TEXT, –

– Foreign key referencing ‘events.id’

subject_id TEXT, –

– Subject entity ID in the triplet

predicate TEXT –

– Predicate describing relation or action

)

“””)

# Commit all changes to the in-memory database

db_conn.commit()

这样就完成了整个引入管道。我们构建了一个系统，可以从原始文本中自动创建一个干净、结构化和时间感知的知识库。下一步也是最后一步是获取这些丰富的数据并构建我们的知识图谱。

组装时态知识图谱

到目前为止，我们已经成功地将原始、混乱的成绩单转换为干净、结构化和时间感知的事实集合。

我们已成功实现：

1. 提取：提取语句、日期和三元组；
2. 解决方法：清理重复实体；
3. 无效：更新不再正确的事实；

现在，是时候获取这些最终的高质量数据并构建我们的知识图谱了。该图将是我们的检索代理将查询以回答用户问题的“大脑”。

图表是此类数据的完美结构，因为它都是关于连接的。

构建和测试多步骤检索代理

我们已经成功地构建了我们的“智能数据库”，这是一个丰富、干净且具有时间感知能力的知识图谱。现在回报来了，构建一个智能代理，可以与此图进行对话以回答复杂的问题。

一个简单的 RAG 系统可能会找到一个相关事实并使用它来回答一个问题。但是，如果答案需要连接多条信息怎么办？

简单的检索系统无法回答这个问题。它需要一个多步骤的过程：

1. 查找有关2016年AMD和数据中心的事实；
2. 查找有关2017年AMD和数据中心的事实；
3. 比较两年的结果；
4. 综合最终摘要；

这就是我们的多步骤检索代理将要做的事情。我们将使用 LangGraph 构建它，它将包含三个关键组件：一个规划器、一组工具和一个 Orchestrator。

在我们的主代理开始工作之前，我们希望它思考并制定一个高层次的计划。这使得代理更加可靠和专注。规划器是一个简单的 LLM 链，其唯一工作是将用户的问题分解为一系列可作的步骤。

代理成功地按照计划进行作，用不同的日期范围调用其工具两次，然后将结果综合成一个完美的比较答案。这证明了将时间知识图谱与多步骤代理相结合的力量。

我们下一步能做什么？

我们构建了一个强大的原型，演示了如何创建一个知识库，它不仅仅是一个静态库，而是一个动态系统，它了解事实如何随时间演变。

现在我们的代理工作了，下一个关键问题是：“它工作得如何？ 回答这个问题需要一个正式的评估过程。主要有三种方法：

1. 黄金答案（黄金标准）：您创建一组测试问题，并让人类专家编写完美的答案。然后，您将代理的输出与这些“黄金”答案进行比较。这是最准确的方法，但速度慢且成本高昂。
2. LLM-as-Judge（可扩展的方法）：您使用强大的LLM（如GPT-4）充当“法官”。它对代理的答案的正确性和相关性进行评分。这既快速又便宜，非常适合快速测试和迭代。
3. 人工反馈（真实世界测试）：部署代理后，您可以添加简单的反馈按钮（如竖起大拇指/竖起大拇指）以让用户对答案进行评分。这告诉您您的代理对实际任务有多大用处。