Lakeflow Declarative Pipelines – Delta Live Tables (DLT)
Delta Live Tables (DLT) is the former name of what Databricks now calls Lakeflow Declarative Pipelines.
In other words: Lakeflow Declarative Pipelines is the current evolution and product name of DLT.
Lakeflow Declarative Pipelines is a data-pipeline framework in Databricks for defining, executing, and maintaining data processing pipelines using a declarative programming model.
The key idea is:
You define the desired state and data relationships of your datasets, rather than explicitly controlling every execution step.
你描述“数据应该是什么样、数据之间是什么关系、数据质量应该满足什么条件”，而不是手工控制每一个执行步骤。
A Declarative Pipeline can contain:
- data sources
- data flows
- streaming tables
- materialized views
- transformations
- data-quality rules
- dependencies between datasets
- pipeline configuration
- operational and monitoring information
The pipeline engine uses the definitions to construct a dependency graph / DAG and determine how the pipeline should be executed.
In current Databricks terminology, the technology formerly known as Delta Live Tables (DLT) has evolved into Lakeflow Declarative Pipelines. The underlying open framework is also referred to as Apache Spark Declarative Pipelines (SDP).
Our Data platform:
ADLS
↓
Bronze
↓
Silver
↓
GoldTraditional data engineering flow looks this:
ADF Trigger
↓
Databricks Job
↓
Notebook 1
↓
Notebook 2
↓
Notebook 3
↓
Data Quality Job
↓
Aggregation Job
↓
Power BI Refresh
Data Engineer have to consider and orchestrate
When does it run? 什么时候运行？
What are the dependencies? 谁依赖谁？
What is the execution order? 谁先执行？
How are failures handled? 失败以后怎么办？
How do we handle re-runs / retries? 重新执行怎么办？
How is incremental processing handled? 如何处理增量？
How do we avoid duplicate data? 如何避免重复？
How is data quality enforced/validated? 如何做数据质量？
How is the pipeline monitored? 如何监控？Declarative Pipeline does this:
Bronze = source data
Silver = cleaned Bronze data
Gold = business aggregation of Silver
------------------------
Silver depends on Bronze
Gold depends on Silver
-----------------------
Bronze
↓
Silver
↓
GoldThis is called Dependency Graph.
The important architectural shift is: Execution logic becomes separated from data logic. 把 Data Logic 和 Execution Logic 分离。
Your code primarily describes:
(Data Engineers are more focused on: Data Engineer 更关注)
- What data should exist? 数据是什么？
- What transformation defines it? 数据怎么转换？
- What quality rules must it satisfy? 数据质量要求是什么？
- What datasets does it depend on? 数据依赖谁？
The platform can then manage much of (而 Pipeline Framework 更多负责):
- Execution 怎么执行
- Dependency resolution 依赖怎么解析
- Incremental processing 怎么增量更新
- Pipeline updates
- Monitoring 怎么监控
- Data quality metrics
- Recovery 怎么处理失败
- Parallel execution 哪些任务可以并行
Declarative Pipeline
│
┌────────────────┼────────────────┐
│ │ │
↓ ↓ ↓
Ingestion Transformation Data Quality
│ │ │
Auto Loader Spark SQL Expectations
│ │ │
↓ ↓ ↓
Bronze ───────→ Silver ─────→ Gold
│
↓
Delta Lake
│
↓
Unity CatalogWhat exactly is the core problem that Declarative Pipelines (DP) aim to solve?
Assuming, a Bank, has those everyday:
Customer
Account
Transaction
Payment
LoanThey come from
Oracle
SQL Server
Kafka
ADLS
REST APIData Engineer should create:
Sources
│
┌───────────┼───────────┐
↓ ↓ ↓
Oracle Kafka ADLS
│ │ │
└───────────┼───────────┘
↓
Bronze
↓
Silver
↓
Gold
↓
Power BIIf we do not have a good pipeline framwork, data engineer may create
ADF Pipeline 1
ADF Pipeline 2
ADF Pipeline 3
Databricks Job 1
Databricks Job 2
Databricks Job 3
Notebook 1
Notebook 2
Notebook 3
DQ Job
CDC Jobmaintaining
dependencies
schedules
retries
checkpoints
parameters
error handling
monitoringOne of the primary goals of Declarative Pipelines (DP) is to shift more of this complexity onto the pipeline framework.”
Key foundational concepts of Declarative Pipelines
Declarative Pipelines
│
├── Pipeline
│
├── Dataset
│ ├── Streaming Table
│ ├── Materialized View
│ └── Temporary View
│
├── Flow
│
├── Dependency Graph
│
├── Incremental Processing
│
├── Data Quality
│ └── Expectations
│
├── Change Data Capture
│ └── AUTO CDC
│
├── Batch Processing
│
├── Streaming Processing
│
├── Monitoring
│ └── Event Log
│
├── Unity Catalog
│
└── Deployment / CI-CDPipeline / Dataset / Flow
| Concept | Professional Meaning | 人话 |
|---|---|---|
| Pipeline | Application / execution boundary | 整个工厂 |
| Dataset | Data object produced/maintained | 产品 |
| Flow | Data movement / processing path | 传送带 |
| Dependency | Relationship between datasets | 谁必须等谁 |
| Pipeline Engine | Executes the declared pipeline | 工厂管理系统 |
Declarative Pipeline Python API code implement
# ============================================================
# Import the modern Declarative Pipelines API
# ============================================================
from pyspark import pipelines as dp
# ============================================================
# BRONZE DATASET
# ============================================================
#
# Purpose:
# Define the Bronze dataset.
#
# Important:
# We are DECLARING what this dataset should contain.
# We are not manually starting a streaming query.
# ============================================================
@dp.table
def customer_bronze():
return (
spark.readStream
# ----------------------------------------------------
# Auto Loader
#
# Purpose:
# Incrementally discover new files arriving in ADLS.
# ----------------------------------------------------
.format("cloudFiles")
# ----------------------------------------------------
# Source format
# ----------------------------------------------------
.option("cloudFiles.format", "json")
# ----------------------------------------------------
# Source location
# ----------------------------------------------------
.load("/mnt/raw/customer")
)
# ============================================================
# SILVER DATASET
# ============================================================
#
# Purpose:
# Clean and standardize Bronze customer data.
#
# Dependency:
#
# customer_bronze
# ↓
# customer_silver
#
# The dependency is expressed by reading customer_bronze.
# ============================================================
@dp.table
def customer_silver():
return (
spark.readStream
# ----------------------------------------------------
# Read the Bronze dataset.
#
# This establishes the logical dependency:
#
# customer_silver
# ↓
# customer_bronze
# ----------------------------------------------------
.table("customer_bronze")
# ----------------------------------------------------
# Data quality / business rule:
#
# A customer record without customer_id is not useful
# to downstream business processing.
# ----------------------------------------------------
.filter("customer_id IS NOT NULL")
# ----------------------------------------------------
# Keep only columns required by the Silver layer.
# ----------------------------------------------------
.select(
"customer_id",
"name",
"email",
"updated_at"
)
)Step by Step build a Complete Lakeflow Declarative Pipeline
Customer Data Lakehouse Pipeline — Requirements
1. Project Goal / 项目目标
Build an end-to-end Lakeflow Declarative Pipeline for customer data:
ADLS Landing → Bronze → Data Quality / Routing → Silver → Gold
The pipeline must demonstrate:
- Auto Loader incremental ingestion
- Bronze raw-data preservation
- Data Quality validation
- Invalid-record quarantine
- Severe-invalid record drop with audit logging
- Silver trusted customer data
- Gold business/analytics output
- Declarative Pipeline dependency management
- Incremental processing
- Auditability and traceability
中文
我们建立一个完整的 Lakeflow Declarative Pipeline：
ADLS Landing → Bronze → 数据质量检查/路由 → Silver → Gold
这个实验不是简单地：
读数据 → filter → 写 Silver
而是模拟一个比较接近企业实际的 Data Engineering Pipeline。
重点学习：
- Auto Loader 增量 ingestion
- Bronze 保存原始数据
- Data Quality validation
- Bad data Quarantine
- 严重坏数据 Drop + Audit Log
- Silver Trusted Data
- Gold Analytics Data
- Declarative Pipeline 自动 dependency management
- Incremental processing
- 数据可追溯性和审计
2. Source Data / 数据源
Raw Data:
customer_001.json
customer_002.json
customer_003.json
ADLS Location: abfss://landing@.dfs.core.windows.net/customer/
ADLS
└── landing
└── customer
├── customers_001.json
├── customers_002.json
└── customers_003.jsonOur pipeline will implement a simple Medallion Architecture using Lakeflow Declarative Pipelines.
Azure ADLS
│
│ JSON files
▼
Auto Loader
│
▼
┌───────────────────┐
│ Bronze │
│ customer_bronze │
│ Raw / append │
└─────────┬─────────┘
│
│ transformation
▼
┌───────────────────┐
│ Silver │
│ customer_silver │
│ Cleaned data │
└─────────┬─────────┘
│
│ aggregation
▼
┌───────────────────┐
│ Gold │
│ customer_summary │
│ Business data │
└───────────────────┘Customer Business Columns
| Column | Meaning |
|---|---|
| customer_id | Customer identifier |
| name | Customer name |
| Customer email | |
| status | ACTIVE / INACTIVE |
| country | Customer country |
| updated_at | Source/business update timestamp |
3. Bronze Requirement / Bronze 要求
Bronze 是整个 pipeline 的原始数据入口层。Ingestion 必须使用：Auto Loader. 要能够：support scalable file ingestion
incremental discover new files
process new files
avoid reprocessing already processed files
4. Bronze Must Preserve Raw Data
这是非常重要的一条 requirement。Bronze 不能只保存清洗后的 business columns。Bronze 必须尽可能保留：
Business Data
customer_id
name
email
status
country
updated_atRaw / Audit Metadata
例如：
_raw_record
_ingest_ts
_source_file
_source_file_name
_source_file_modification_time如果适用，也可以考虑：
_source_system
_batch_id
_record_hash5. Important Timestamp Definition
这里要特别区分两个时间, updated_at and _ingest_ts
updated_at: 这是 Source Business Timestamp, 表示源系统认为这条 customer record 什么时候更新。例如： updated_at = 2026-09-10 10:05:00
_ingest_ts: 这是 Data Platform Ingestion Timestamp, 表示这条数据什么时候进入我们的 Data Platform / Pipeline processing。例如：_ingest_ts = 2026-09-10 10:06:13
两者不能混为一谈。
6. Bronze Raw Record Requirement
我们希望 Bronze 能够保存原始 record 的 JSON representation，例如：_raw_record
这样以后：如果一条 record 被 quarantine 或 drop，我们可以看到它进入 pipeline 时的原始状态。这是企业 Data Platform 中非常重要的：Traceability / 可追溯性
7. Data Quality Requirements
这是整个实验最重要的部分。我们定义三个核心 Data Quality rules：
Rule 1 — Customer ID
customer_id must not be NULLRule 2 — Email
Email 必须：
NOT NULL并且应该满足基本 email format。例如：alice@example.com 有效。而：alice@ 无效。
Rule 3 — Status
Status 必须是：ACTIVE 或者：INACTIVE . 其他值都是 invalid。 例如：
ACTIVE → valid
INACTIVE → valid
PENDING → invalid
UNKNOWN → invalid
NULL → invalid8. Data Quality Routing Requirement
我们不是简单地 DROP 所有坏数据。这是这个项目的核心设计。我们把 record 分成三个类别：
Bronze
│
▼
Data Quality
Validation
│
┌────────────┼────────────┐
▼ ▼ ▼
VALID RECOVERABLE SEVERE
│ │ │
▼ ▼ ▼
Silver Quarantine Drop Log9. Routing Decision Matrix
这是目前我们确认的正式 requirement baseline。
| Customer_ID | Status | Action | Destination | |
|---|---|---|---|---|
| Valid | Valid | ACTIVE | ACCEPT | Silver |
| Valid | Valid | INACTIVE | ACCEPT | Silver |
| NULL | Valid | Valid | QUARANTINE | Quarantine |
| Valid | NULL | Valid | QUARANTINE | Quarantine |
| Valid | Valid | Invalid | QUARANTINE | Quarantine |
| NULL | NULL | Valid | QUARANTINE | Quarantine |
| NULL | Valid | Invalid | QUARANTINE | Quarantine |
| Valid | NULL | Invalid | QUARANTINE | Quarantine |
| NULL | NULL | Invalid | DROP | Drop Log |
10. Quarantine Requirement
如果 record 不符合正常 Silver 数据质量要求，但是没有达到 Severe Drop 条件：Quarantine
例如：
customer_id = NULL
email = bob@example.com
status = ACTIVE不能进入 Silver。但是：不能直接删除。应该进入：
customer_quarantine11. Quarantine Must Preserve Original Record
Quarantine 至少应该包含：Original Business Data
customer_id
name
email
status
country
updated_atOriginal Record
_raw_recordSource Information
_source_file
_source_file_nameIngestion Information
_ingest_tsData Quality Information
_dq_customer_id_valid
_dq_email_valid
_dq_status_valid以及：
_dq_reason例如：
CUSTOMER_ID_NULL或者：
EMAIL_NULL或者：
INVALID_STATUS多个问题：
CUSTOMER_ID_NULL;EMAIL_NULL12. Quarantine Timestamp
需要记录：_quarantine_ts . 表示：这条 record 被判定并进入 quarantine 的时间。
13. Drop Requirement
最严重的情况：
customer_id = NULL
email = NULL
status = INVALID这条 record：不进入 Silver, 不作为正常 Quarantine Record. 而是：DROP
但是：Drop ≠ 数据完全消失,
这是非常重要的概念。真正的设计是：
Bronze
│
└── severe invalid
│
▼
Drop Audit Log所以我们实际上是：从 business processing path 中 drop，但从 audit trail 中保留。
14. Drop Log Requirement
建立：customer_drop_log
至少保存：
Original Data
customer_id
name
email
status
country
updated_atRaw Record
_raw_recordSource Metadata
_source_file
_source_file_name
_source_file_modification_timeIngestion Information
_ingest_tsDrop Information
_drop_ts
_drop_reason例如：
_drop_reason =
CUSTOMER_ID_NULL;EMAIL_NULL;INVALID_STATUS这样以后可以回答：Why was this customer record dropped?
以及：What exactly was the original record?
15. Silver Requirement
Silver 是：Trusted / Validated Customer Data
只有：
customer_id valid
AND
email valid
AND
status ∈ ('ACTIVE', 'INACTIVE')的 record 才能进入 Silver。因此：
Bronze
│
├── valid ─────────────→ Silver
│
├── invalid/recoverable → Quarantine
│
└── severe invalid ────→ Drop LogSilver 不应该包含：
- NULL customer_id
- NULL email
- invalid email
- invalid status
16. Silver Transformation Requirement
Silver 除了 DQ filtering，还可以做一些正常的 standardization，例如：
trim(customer_id)
trim(name)
lower(email)
upper(status)例如：
" alice@example.com "standardize 成：
"alice@example.com"但是：Bronze 不修改原始数据。这是：
Bronze = Preserve
Silver = Clean / Standardize / Trust17. Gold Requirement
Gold 是面向：Business / Analytics Consumption. 我们需要一个简单但有意义的 Gold dataset。
例如：
customer_summary按：
status统计 customer 数量：
status customer_count
---------------------------
ACTIVE 5
INACTIVE 1也可以进一步按：
country
status进行 aggregation。
例如：
country status customer_count
---------------------------------------
Canada ACTIVE 4
Canada INACTIVE 0
USA INACTIVE 1
UK ACTIVE 1Gold 使用：Materialized View 来体现：
Silver
↓
Gold Materialized View18. Final Pipeline Architecture
ADLS
│
│
Auto Loader
│
▼
┌───────────┐
│ Bronze │
│ Raw Data │
│ + Metadata│
└─────┬─────┘
│
▼
Data Quality Rules
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
VALID RECOVERABLE SEVERE
│ │ │
▼ ▼ ▼
Silver Quarantine Drop Log
│
│
▼
Gold Materialized
View19. Final Dataset List
整个 Pipeline 最终应该至少有 4 个主要数据集：
1. customer_bronze
2. customer_silver
3. customer_quarantine
4. customer_drop_log
5. customer_summary ← Gold也就是说：
Bronze
│
├──────────────→ Silver
│ │
│ ▼
│ Gold
│
├──────────────→ Quarantine
│
└──────────────→ Drop Log20. Declarative Pipeline Requirement
我们不希望自己手工 orchestration, 而是 Declare the datasets and their relationships.
Silver reads Bronze
Gold reads Silver
Quarantine reads Bronze
Drop Log reads BronzePipeline framework 根据这些 dependencies 自动建立：
Dependency Graph然后负责：
- execution
- dependency resolution
- orchestration
- incremental processing
- checkpointing
- pipeline updates
- table maintenance
我们的代码主要负责：定义“数据应该是什么样子，以及它从哪里来”。
22. Incremental Processing Requirement
实验必须体现：First Run 例如：customers_001.json ; customers_002.json, 进入 pipeline。
Later 再增加：customers_003.json
Pipeline update 后：只处理新的 input data，而不是每次重新从头读取所有文件。
最终我们观察：
Bronze
Silver
Quarantine
Drop Log
Gold23. Data Lineage / Traceability Requirement
我们最终应该能够回答：Where did this Silver record come from?
Silver
↓
Bronze
↓
ADLS fileWhy was this record quarantined?
Quarantine
↓
DQ reason
↓
Original record
↓
Source fileWhy was this record dropped?
Drop Log
↓
Drop reason
↓
Original record
↓
Source file
↓
Ingestion timestamp这就是：
Data Lineage + Auditability + Traceability
25. Coding Order / 我们之后的 Coding 顺序
Step 1: Create Bronze
ADLS
↓
Auto Loader
↓
customer_bronzeStep 2: Validate Bronze Data
建立：
customer_id validation
email validation
status validationStep 3: Classify / Route Records
VALID
QUARANTINE
DROPStep 4: Create Quarantine
customer_quarantineStep 5: Create Drop Audit Log
customer_drop_logStep 6: Create Silver
customer_silverStep 7: Create Gold
customer_summaryStep 8: Test Incremental Processing
增加新的 JSON：
customers_003.jsonStep 9: Review Pipeline Graph / Data Quality / Audit
Test Data
we want to demonstrate incremental ingestion.
# File 1 — customers_001.json
[
{
"customer_id": "C001",
"name": "Alice",
"email": "alice@example.com",
"status": "ACTIVE",
"country": "Canada",
"updated_at": "2026-09-10T09:00:00"
},
{
"customer_id": "C002",
"name": "Bob",
"email": "bob@example.com",
"status": "ACTIVE",
"country": "Canada",
"updated_at": "2026-09-10T09:05:00"
},
{
"customer_id": "C003",
"name": "Charlie",
"email": "charlie@example.com",
"status": "INACTIVE",
"country": "USA",
"updated_at": "2026-09-10T09:10:00"
}
]
# File 2 — customers_002.json
[
{
"customer_id": "C004",
"name": "David",
"email": "david@example.com",
"status": "ACTIVE",
"country": "Canada",
"updated_at": "2026-09-10T10:00:00"
},
{
"customer_id": "C005",
"name": "Emma",
"email": "emma@example.com",
"status": "ACTIVE",
"country": "UK",
"updated_at": "2026-09-10T10:05:00"
}
]URL: abfss://landing@storageaccount.dfs.core.windows.net/customer/
Pipeline source code will like:
customer_pipeline/
│
├── customer_pipeline.py
│
└── configuration
1: Create Bronze (Simple, no for product)
aim: Incrementally load JSON files from ADLS into a Bronze Streaming Table.
# 简单实现， 其实在实际项目不会这么简单
from pyspark import pipelines as dp
@dp.table
def customer_bronze(): # create a table called "customer_bronze"
# The decorator declares that the function defines a pipeline-managed dataset.
# In this context, it creates a streaming table.
return (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.load("abfss://landing@myaccount.dfs.core.windows.net/customer/")
)we do not do ” create table ……” then “writeStream.start()” ,
Declarative Pipeline WILL TAKE OVER this job
The pipeline framework manages:
- execution
- checkpointing
- dependency resolution
- incremental processing
- scheduling
- table maintenance
- orchestration
The developer primarily declares: what dataset should exist and how it is derived.
2: Create Silver (Simple, not for product)
@dp.table
@dp.expect( # Data Quality Expectation
"valid_customer_id",
"customer_id IS NOT NULL" # customer_id cannot null
)
@dp.expect(
"valid_email",
"email IS NOT NULL AND email RLIKE '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'"
# email cannot be null and it should be in good form
)
@dp.expect(
"valid_status",
"status IN ('ACTIVE', 'INACTIVE')" # status only allow ACTIVE or INACTIVE
)
def customer_silver():
return (
spark.readStream
.table("customer_bronze")
.filter("customer_id IS NOT NULL")
.select(
"customer_id",
"name",
"email",
"status",
"country",
"updated_at"
)
)Product grade, end- to-end code implement
test data
abfss://landing@<storage-account>.dfs.core.windows.net/customer/
customers_001.json
customers_002.json
[
{
"customer_id": "C001",
"name": "Alice",
"email": "alice@example.com",
"status": "ACTIVE",
"country": "Canada",
"updated_at": "2026-09-10T09:00:00"
},
{
"customer_id": "C002",
"name": "Bob",
"email": "bob@example.com",
"status": "ACTIVE",
"country": "Canada",
"updated_at": "2026-09-10T09:05:00"
},
{
"customer_id": "C003",
"name": "Charlie",
"email": "charlie@example.com",
"status": "INACTIVE",
"country": "USA",
"updated_at": "2026-09-10T09:10:00"
}
]Step 1 — Create Bronze
Our first pipeline layer is:
ADLS Landing
│
│ Auto Loader
▼
customer_bronzeBronze’s job is not to clean the data. Its job is: Ingest and preserve the source data faithfully, together with ingestion/source metadata.
First Bronze implementation
from pyspark import pipelines as dp
@dp.table
def customer_bronze(): # create a table called "customer_bronze"
return (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.load(
"abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
)
)What does this code actually produce? it creates “customer_bronze”
Table customer_bronze:
customer_id
name
email
status
country
updated_at
# we're missing
├── _raw_record
├── _ingest_ts
├── _source_file
├── _source_file_name
└── _source_file_modification_time
这些是我们之前 Requirement 中要求的 audit information。只有改成下面的现在我们把代码升级， 满足requirement 的要求。
from pyspark import pipelines as dp
from pyspark.sql.functions import (
current_timestamp,
col,
struct,
to_json
)
@dp.table(
comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():
return (
spark.readStream
# ---------------------------------------------------------
# Use Databricks Auto Loader as the streaming file source.
# Auto Loader incrementally discovers new files arriving
# in the ADLS landing folder.
# ---------------------------------------------------------
.format("cloudFiles")
# ---------------------------------------------------------
# The source files in our experiment are JSON files.
# ---------------------------------------------------------
.option("cloudFiles.format", "json")
# ---------------------------------------------------------
# Load files from the ADLS landing zone.
#
# Replace <storage-account> with the real storage account.
# ---------------------------------------------------------
.load(
"abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
)
# ---------------------------------------------------------
# Add platform ingestion timestamp.
#
# This is NOT the source business timestamp.
# updated_at = source/business time
# _ingest_ts = data-platform ingestion/processing time
# ---------------------------------------------------------
.withColumn(
"_ingest_ts",
current_timestamp()
)
# ---------------------------------------------------------
# Capture source file metadata.
#
# _metadata is provided by Auto Loader.
# We preserve the file path and modification time so that
# downstream DQ issues can be traced back to the source file.
# ---------------------------------------------------------
.withColumn(
"_source_file",
col("_metadata.file_path")
)
.withColumn(
"_source_file_modification_time",
col("_metadata.file_modification_time")
)
# ---------------------------------------------------------
# Keep the raw source record as JSON.
#
# This gives us a representation of the record as it exists
# at the Bronze stage before Silver transformations.
#
# This will later be extremely useful for:
# - Quarantine
# - Drop Audit Log
# - Troubleshooting
# - Data lineage / traceability
# ---------------------------------------------------------
.withColumn(
"_raw_record",
to_json(
struct(
"customer_id",
"name",
"email",
"status",
"country",
"updated_at"
)
)
)
# ---------------------------------------------------------
# Select the columns that belong to the Bronze contract.
#
# Bronze preserves business data plus source/audit metadata.
# ---------------------------------------------------------
.select(
"customer_id",
"name",
"email",
"status",
"country",
"updated_at",
"_raw_record",
"_ingest_ts",
"_source_file",
"_source_file_modification_time"
)
)三个时间的区别 (以后你面试很可能会被问到)
| Field | Meaning |
|---|---|
updated_at | Source business timestamp |
_source_file_modification_time | Source file timestamp |
_ingest_ts | Platform ingestion timestamp |
为什么需要 _raw_record？ 这是我们后面设计：
Quarantine
Drop LogCurrent Bronze Architecture
现在我们已经从：
ADLS
↓
Auto Loader
↓
Bronze升级成：
ADLS
│
▼
Auto Loader
│
▼
┌──────────────────┐
│ customer_bronze │
├──────────────────┤
│ customer_id │
│ name │
│ email │
│ status │
│ country │
│ updated_at │
│ │
│ _raw_record │
│ _ingest_ts │
│ _source_file │
│ _source_file_... │
└──────────────────┘What should we see after the first run? 如果 source 中：
C001
C002
C003
C004
C005那么 Bronze 应该有：5 records. 例如：
| customer_id | status | _source_file | _ingest_ts | |
|---|---|---|---|---|
| C001 | alice@example.com | ACTIVE | customers_001.json | … |
| C002 | bob@example.com | ACTIVE | customers_001.json | … |
| C003 | charlie@example.com | INACTIVE | customers_001.json | … |
| C004 | david@example.com | ACTIVE | customers_002.json | … |
| C005 | emma@example.com | ACTIVE | customers_002.json | … |
注意：我们还没有进行任何 Business Data Quality filtering。
所以把代码补完整：
from pyspark import pipelines as dp
from pyspark.sql.functions import (
current_timestamp,
col,
struct,
to_json
)
@dp.table(
comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():
return (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.load(
"abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
)
# Platform ingestion timestamp
.withColumn(
"_ingest_ts",
current_timestamp()
)
# Source file metadata
.withColumn(
"_source_file",
col("_metadata.file_path")
)
.withColumn(
"_source_file_name",
col("_metadata.file_name")
)
.withColumn(
"_source_file_modification_time",
col("_metadata.file_modification_time")
)
# Serialized representation of the parsed business record
.withColumn(
"_raw_record",
to_json(
struct(
"customer_id",
"name",
"email",
"status",
"country",
"updated_at"
)
)
)
.select(
"customer_id",
"name",
"email",
"status",
"country",
"updated_at",
"_raw_record",
"_ingest_ts",
"_source_file",
"_source_file_name",
"_source_file_modification_time"
)
)Step 2 — Data Quality Validation
数据质量验证
from pyspark import pipelines as dp
from pyspark.sql.functions import (
col,
when,
concat_ws,
array,
lit
)
@dp.temporary_view
def customer_validation():
return (
spark.readStream
.table("customer_bronze")
# ==============================
# DQ Rule 1: Customer ID
# ==============================
.withColumn(
"_dq_customer_id_valid",
col("customer_id").isNotNull()
)
# ==============================
# DQ Rule 2: Email
# ==============================
.withColumn(
"_dq_email_valid",
(
col("email").isNotNull()
&
col("email").rlike(
r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
)
)
)
# ==============================
# DQ Rule 3: Status
# ==============================
.withColumn(
"_dq_status_valid",
col("status").isin("ACTIVE", "INACTIVE")
)
# ==============================
# Overall validation
# ==============================
.withColumn(
"_dq_all_valid",
(
col("_dq_customer_id_valid")
&
col("_dq_email_valid")
&
col("_dq_status_valid")
)
)
# ==============================
# DQ reason
# ==============================
.withColumn(
"_dq_reason",
concat_ws(
";",
array(
when(
col("customer_id").isNull(),
lit("CUSTOMER_ID_NULL")
),
when(
col("email").isNull(),
lit("EMAIL_NULL")
),
when(
col("email").isNotNull()
&
~col("email").rlike(
r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
),
lit("INVALID_EMAIL_FORMAT")
),
when(
~col("status").isin("ACTIVE", "INACTIVE"),
lit("INVALID_STATUS")
)
)
)
)
# ==============================
# Routing decision
# ==============================
.withColumn(
"_dq_action",
when(
col("_dq_all_valid"),
lit("ACCEPT")
)
.when(
(~col("_dq_customer_id_valid"))
&
(~col("_dq_email_valid"))
&
(~col("_dq_status_valid")),
lit("DROP")
)
.otherwise(
lit("QUARANTINE")
)
)
)Step 3 — Routing into Silver / Quarantine / Drop Log
Silver 的原则：Only trusted, validated customer records enter Silver.
from pyspark import pipelines as dp
from pyspark.sql.functions import (
col,
when,
concat_ws,
array,
lit,
trim,
lower,
upper,
current_timestamp
)
# ============================================================
# 1. Validation and Routing
# ============================================================
@dp.temporary_view
def customer_validation():
return (
spark.readStream
.table("customer_bronze")
# ------------------------------------------------
# DQ Rule 1: customer_id must not be NULL
# ------------------------------------------------
.withColumn(
"_dq_customer_id_valid",
col("customer_id").isNotNull()
)
# ------------------------------------------------
# DQ Rule 2:
# email must not be NULL and must match
# the basic email format
# ------------------------------------------------
.withColumn(
"_dq_email_valid",
(
col("email").isNotNull()
&
col("email").rlike(
r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
)
)
)
# ------------------------------------------------
# DQ Rule 3:
# status must be ACTIVE or INACTIVE
# ------------------------------------------------
.withColumn(
"_dq_status_valid",
(
col("status").isNotNull()
&
col("status").isin(
"ACTIVE",
"INACTIVE"
)
)
)
# ------------------------------------------------
# Overall DQ result
# ------------------------------------------------
.withColumn(
"_dq_all_valid",
(
col("_dq_customer_id_valid")
&
col("_dq_email_valid")
&
col("_dq_status_valid")
)
)
# ------------------------------------------------
# DQ Reason
#
# Multiple reasons are concatenated:
#
# CUSTOMER_ID_NULL;EMAIL_NULL
# ------------------------------------------------
.withColumn(
"_dq_reason",
concat_ws(
";",
array(
when(
col("customer_id").isNull(),
lit("CUSTOMER_ID_NULL")
),
when(
col("email").isNull(),
lit("EMAIL_NULL")
),
when(
col("email").isNotNull()
&
~col("email").rlike(
r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
),
lit("INVALID_EMAIL_FORMAT")
),
when(
col("status").isNull()
|
~col("status").isin(
"ACTIVE",
"INACTIVE"
),
lit("INVALID_STATUS")
)
)
)
)
# ------------------------------------------------
# Routing Decision
#
# ACCEPT:
# All three rules pass
#
# DROP:
# All three rules fail
#
# QUARANTINE:
# Everything else
# ------------------------------------------------
.withColumn(
"_dq_action",
when(
col("_dq_all_valid"),
lit("ACCEPT")
)
.when(
(~col("_dq_customer_id_valid"))
&
(~col("_dq_email_valid"))
&
(~col("_dq_status_valid")),
lit("DROP")
)
.otherwise(
lit("QUARANTINE")
)
)
)
# ============================================================
# 2. Silver
# ============================================================
@dp.table(
comment="Trusted customer records that passed all core data quality rules."
)
def customer_silver():
return (
spark.readStream
.table("customer_validation")
# Only fully valid records
.filter(
col("_dq_action") == "ACCEPT"
)
# Standardize trusted business data
.select(
trim(
col("customer_id")
).alias("customer_id"),
trim(
col("name")
).alias("name"),
lower(
trim(
col("email")
)
).alias("email"),
upper(
trim(
col("status")
)
).alias("status"),
trim(
col("country")
).alias("country"),
col("updated_at"),
# Audit / lineage metadata
col("_raw_record"),
col("_ingest_ts"),
col("_source_file"),
col("_source_file_name"),
col("_source_file_modification_time")
)
)
# ============================================================
# 3. Quarantine
# ============================================================
@dp.table(
comment="Customer records that failed one or more recoverable data quality rules."
)
def customer_quarantine():
return (
spark.readStream
.table("customer_validation")
# Recoverable DQ failures
.filter(
col("_dq_action") == "QUARANTINE"
)
.select(
# Original business data
"customer_id",
"name",
"email",
"status",
"country",
"updated_at",
# Raw / lineage information
"_raw_record",
"_ingest_ts",
"_source_file",
"_source_file_name",
"_source_file_modification_time",
# DQ results
"_dq_customer_id_valid",
"_dq_email_valid",
"_dq_status_valid",
"_dq_reason",
# Quarantine audit timestamp
current_timestamp().alias(
"_quarantine_ts"
)
)
)
# ============================================================
# 4. Drop Log
# ============================================================
@dp.table(
comment="Audit log for severely invalid customer records excluded from normal processing."
)
def customer_drop_log():
return (
spark.readStream
.table("customer_validation")
# Severe DQ failures
.filter(
col("_dq_action") == "DROP"
)
.select(
# Original business data
"customer_id",
"name",
"email",
"status",
"country",
"updated_at",
# Raw / lineage information
"_raw_record",
"_ingest_ts",
"_source_file",
"_source_file_name",
"_source_file_modification_time",
# DQ results
"_dq_customer_id_valid",
"_dq_email_valid",
"_dq_status_valid",
# Audit reason
col("_dq_reason").alias(
"_drop_reason"
),
# Drop audit timestamp
current_timestamp().alias(
"_drop_ts"
)
)
)这四个 Dataset 的关系
customer_bronze
│
│
▼
┌────────────────────┐
│ customer_validation│
│ │
│ DQ Rules │
│ + Reason │
│ + Routing │
└─────────┬──────────┘
│
┌────────────┼────────────┐
│ │ │
ACCEPT QUARANTINE DROP
│ │ │
▼ ▼ ▼
customer_silver customer_ customer_
quarantine drop_log最终 Routing Logic
| customer_id | status | _dq_action | 去哪里 | |
|---|---|---|---|---|
| valid | valid | ACTIVE | ACCEPT | Silver |
| valid | valid | INACTIVE | ACCEPT | Silver |
| NULL | valid | valid | QUARANTINE | Quarantine |
| valid | NULL | valid | QUARANTINE | Quarantine |
| valid | invalid format | valid | QUARANTINE | Quarantine |
| valid | valid | invalid | QUARANTINE | Quarantine |
| NULL | NULL | valid | QUARANTINE | Quarantine |
| NULL | valid | invalid | QUARANTINE | Quarantine |
| valid | NULL | invalid | QUARANTINE | Quarantine |
| NULL | NULL | invalid | DROP | Drop Log |
Step 4 — Gold: customer_summary
from pyspark import pipelines as dp
@dp.materialized_view(
comment="Gold customer summary by customer status."
)
def customer_summary():
return (
spark.read
.table("customer_silver")
.groupBy("status")
.count()
.withColumnRenamed(
"count",
"customer_count"
)
)完整架构：
ADLS Landing
│
▼
┌─────────────┐
│ Auto Loader │
└──────┬──────┘
│
▼
┌───────────────┐
│ Bronze │
│ Raw + Metadata│
└───────┬───────┘
│
▼
┌─────────────────────┐
│ Customer Validation │
│ │
│ DQ Rules │
│ DQ Reason │
│ Routing │
└─────────┬───────────┘
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
ACCEPT QUARANTINE DROP
│ │ │
▼ ▼ ▼
Silver Quarantine Drop Log
│
▼
Materialized View
│
▼
Gold
│
▼
customer_summaryLakeflow Declarative Pipeline
我们现在写的是：
@dp.table
def customer_bronze():
...然后：
@dp.temporary_view
def customer_validation():
return spark.readStream.table("customer_bronze")然后：
@dp.table
def customer_silver():
return spark.readStream.table("customer_validation")最后：
@dp.materialized_view
def customer_summary():
return spark.read.table("customer_silver")Streaming Table vs Materialized View
| Streaming Table | Materialized View | |
|---|---|---|
| 典型用途 | incremental row processing | derived / aggregated result |
| 数据特点 | 持续到来的数据 | 查询计算结果 |
| 常见操作 | filter / transform / enrich | aggregation / business metrics |
| API | @dp.table | @dp.materialized_view |
| 我们的例子 | customer_silver | customer_summary |
Step 6 — Put the Code into a Lakeflow Pipeline
现在进入真正的 hands-on 部分：
我们已经写完 Dataset definitions。下一步就是把这些 definitions 放进一个 Lakeflow Pipeline，让 Databricks 真正执行它。
这里先只做 Pipeline structure + code organization，暂时不深入 Deployment / Schedule / Production settings。
先把代码组织起来
最简单的方式是一个 Python pipeline source file，例如：
customer_pipeline/
│
└── customer_pipeline.py里面包含我们目前的全部 Dataset definitions：
customer_pipeline.py
│
├── customer_bronze
│
├── customer_validation
│
├── customer_silver
│
├── customer_quarantine
│
├── customer_drop_log
│
└── customer_summaryData Federation
Data Federation lets Databricks/Unity Catalog access and query data that remains in its original system, without first copying all of that data into Delta tables in Databricks.
Suppose I have multiple data:
on-prime: SQL Server, MySQL, Oracle;
Clould Azure SQL DB;
Cloud files on ADLS, format in Delta, Clould files on Azure Blob
My target environment is:
Azure Databricks
│
Unity Catalog
│
┌──────────┴──────────┐
│ Data Federation │
└──────────┬──────────┘
│
┌─────────────────┼─────────────────┐
│ │ │
On-Prem Azure Cloud Files
│ │ │
┌─────┼─────┐ Azure SQL DB ADLS Gen2
│ │ │ Delta files
SQL MySQL Oracle Blob Storage
ServerThink of two fundamentally different approaches. Traditional approach — copy the data
SQL Server
│
│ ETL / ADF / Spark
▼
ADLS
│
▼
Delta Lake
│
▼
Unity Catalog
│
▼
Databricks SQL
The data physically moves.
Federation approach, the customer data can remain inside SQL Server.
Unity Catalog
│
Foreign Catalog
│
┌────────────────┼────────────────┐
│ │ │
SQL Server Oracle MySQL
│ │ │
└────────────────┼────────────────┘
│
Query FederationThe data stays where it is. You can query data where it lives.
Databricks provides a governed way to access it.
SELECT *
FROM sqlserver_catalog.sales.customers;The customer data can remain inside SQL Server.
The Unity Catalog mental model
Federation through this hierarchy:
Unity Catalog
│
├── Metastore
│
├── Catalog
│
│ ├── Schema
│ │
│ │ └── Table
│ │
│ └── ...
│
├── Connections
│
├── Foreign Catalogs
│
├── Storage Credentials
│
├── External Locations
│
└── VolumesFor Federation, pay particular attention to:
Connection
↓
Foreign Catalog
↓
Foreign Schema
↓
Foreign TableConceptually
SQL Server
│
│ connection
▼
Unity Catalog
│
▼
Foreign Catalog
│
├── schema1
│ ├── tableA
│ └── tableB
│
└── schema2
└── tableCHands-on learning lab
We’ll create:
LAB
│
├── SQL Server
│
├── MySQL
│
├── Oracle
│
├── Azure SQL Database
│
├── ADLS Gen2
│ └── Delta
│
└── Azure Blob
└── CSV/ParquetAnd connect these to:
Azure Databricks
│
▼
Unity Catalog
│
├── sqlserver_federated
├── mysql_federated
├── oracle_federated
├── azuresql_federated
│
├── catalog_ADLS_Internal
│
└── blob_externalThen we’ll run queries combining the data.
Lab scenario
Let’s create a realistic business scenario. Suppose the company has:
SQL Server
Customer master:Customer
- CustomerID
- CustomerName
- Region
- CustomerType
Oracle
Orders:Orders
- OrderID
- CustomerID
- OrderDate
- Amount
MySQL
Products:
- ProductID
- ProductName
- Category
- Price
Azure SQL
Customer transactions:
Transaction
- TransactionID
- CustomerID
- ProductID
- Amount
- TransactionDate
ADLS Delta
Historical sales: sales_history
Blob
Marketing CSV: customer_campaign.csv
Now we can ask: Show total customer sales across Oracle historical orders, Azure SQL transactions and ADLS Delta, enriched with customer information from SQL Server.
Step 1 — Prepare Unity Catalog
First verify your current UC environment, ensure we have enable “Unity Catalog”, and you have grant to create catalog, schema, table, external location, Connection
Step 2 — Understand the Connection
This is one of the most important concepts. A Connection represents how Databricks reaches the external database.
# Conceptually:
Unity Catalog
│
▼
Connection
│
├── hostname
├── port
├── database
├── authentication
└── credentials
│
▼
SQL ServerStep 3 — SQL Server Federation
For SQL Server, you create a connection to the SQL Server database. The SQL syntax follows the Unity Catalog federation model.
CREATE CONNECTION sqlserver_prod
TYPE SQLSERVER
OPTIONS (
host '10.100.5.3',
port '1433',
user 'william',
password '123456',
database 'Customer_master_DB'
);
Step 4 — Create the Foreign Catalog
Once the connection exists:
CREATE FOREIGN CATALOG sqlserver_prod_catalog
USING CONNECTION sqlserver_prod;
SHOW CATALOGS;
might show:
catalog_ut_internal
sqlserver_catalog
system
SHOW SCHEMAS IN sqlserver_prod_catalog;
might see
dbo
...
SHOW TABLES IN sqlserver_prod_catalog.dbo;
might see
Customer
CustomerAddress
CustomerTypeYou didn’t copy those tables into Delta. Databricks is exposing the external database through Unity Catalog.
Step 5 — Query the federated table
SELECT *
FROM sqlserver_prod_internal.dbo.Customer
LIMIT 10;Step 6 — MySQL
# dummay
# create CONNECTION
CREATE CONNECTION mysql_prod
TYPE MYSQL
OPTIONS (
host 'mysql-demo.company.com',
port '3306',
user 'dbx_federation',
password 'YourStrongPasswordHere'
);
# create FOREIGN CATALOG
CREATE FOREIGN CATALOG mysql_sales
USING CONNECTION mysql_prod
OPTIONS (
database 'salesdb'
);
# query
SELECT *
FROM mysql_sales.salesdb.customers;
# different Foreign catalog
SELECT *
FROM mysql_sales.salesdb.orders o
JOIN sqlserver_prod_catalog.dbo.Customer sc
ON o.customer_id = sc.customer_id;Step 7 — Oracle
#CREATE CONNECTION
CREATE CONNECTION oracle_prod
TYPE ORACLE
OPTIONS (
host 'oracle-demo.company.com',
port '1521',
user 'federation_user',
password 'YourStrongPasswordHere',
service_name 'ORCL'
);
#Oracle Foreign Catalog
CREATE FOREIGN CATALOG oracle_sales
USING CONNECTION oracle_prod;
#query
SELECT
c.customer_id,
c.customer_name,
SUM(m.amount) AS mysql_sales,
SUM(o.amount) AS oracle_sales
FROM sqlserver_federated.dbo.customers c
LEFT JOIN mysql_sales.salesdb.orders m
ON c.customer_id = m.customer_id
LEFT JOIN oracle_sales.federation_user.orders o
ON c.customer_id = o.customer_id
GROUP BY
c.customer_id,
c.customer_name
ORDER BY
c.customer_id;Step 8 — Azure SQL Database
# Create connection
CREATE CONNECTION azure_sql_prod
TYPE sqlserver
OPTIONS (
host 'myazuresqlserver.database.windows.net',
port '1433',
user 'dbx_federation',
password secret('my-secret-scope', 'azure-sql-password') -- password 'YourPassword'
);
# pay attention
I used --> secret('my-secret-scope', 'azure-sql-password')
# Create catalog
CREATE FOREIGN CATALOG azure_sql_sales
USING CONNECTION azure_sql_prod
OPTIONS (
database 'federation_lab'
);
# Query Azure SQL
SELECT *
FROM azure_sql_sales.dbo.customers;ADLS Gen2 + Delta
Does not “connection” , “Foreign Catalog” !
Create Storage Credential
Create External Location
# Create external location
CREATE EXTERNAL LOCATION adls_federation_location
URL 'abfss://data@mystoragelab.dfs.core.windows.net/federation/'
WITH (STORAGE CREDENTIAL adls_federation_cred);
# Create catallog
# query
SELECT *
FROM catalog_ADLS_Internal.federation_lab.sales_demo;Blob
Does not “connection” , “Foreign Catalog” !
Create Blob External Location
# create EXTERNAL LOCATION
CREATE EXTERNAL LOCATION blob_federation_location
URL 'abfss://files@mybloblab.dfs.core.windows.net/federation/'
WITH (STORAGE CREDENTIAL blob_storage_cred);
# Create External Table for CSV
CREATE TABLE catalog_ADLS_Internal.federation_lab.marketing_campaigns
USING CSV
OPTIONS (
header = 'true',
inferSchema = 'true'
)
LOCATION
'abfss://files@mybloblab.dfs.core.windows.net/federation/marketing/';
# query
SELECT *
FROM catalog_ADLS_Internal.federation_lab.marketing_campaigns;Remember this:
| Source | UC Object | 关键代码 |
|---|---|---|
| SQL Server | Foreign Catalog | CREATE CONNECTION |
| MySQL | Foreign Catalog | CREATE CONNECTION |
| Oracle | Foreign Catalog | CREATE CONNECTION |
| Azure SQL DB | Foreign Catalog | CREATE CONNECTION |
| ADLS Gen2 + Delta | External Table | CREATE EXTERNAL LOCATION |
| Azure Blob | External Table | CREATE EXTERNAL LOCATION |
Databricks MCP Server
A Databricks MCP Server is a Model Context Protocol server that exposes Databricks platform capabilities (Unity Catalog, SQL, Jobs, Clusters, etc.) as MCP tools that AI agents can dynamically discover and call.
CN: Databricks MCP Server 是一个 MCP 服务器，它将 Databricks 平台的能力（Unity Catalog、SQL、Jobs、Clusters 等）暴露为 MCP 工具，让 AI Agent 可以动态发现和调用这些工具。
Databricks provides three types of MCP servers:
CN: Databricks 提供三种类型的 MCP 服务器：
2.1 Managed MCP Servers（托管 MCP 服务器）
EN: Databricks provides the following managed MCP servers that work out of the box:
CN: Databricks 提供以下开箱即用的托管 MCP 服务器：
2.2 Custom MCP Servers（自定义 MCP 服务器）
EN: You can host custom or third-party MCP servers as Databricks Apps. This is useful if you:
- Already have an MCP server you want to deploy
- Want to run a third-party MCP server as a tool source
- Need custom business logic not covered by managed servers
CN: 你可以将自定义或第三方的 MCP 服务器托管为 Databricks Apps。 这在以下情况很有用：
Key Takeaways
MCP testing locally: Claude Desktop Connection
Claude Desktop Connection is the process of configuring Anthropic’s Claude desktop application to communicate with local MCP (Model Context Protocol) servers via stdio (standard input/output) transport.
CN: Claude Desktop 连接是指将 Anthropic 的 Claude 桌面应用程序配置为通过 stdio（标准输入/输出）传输方式与本地 MCP（模型上下文协议）服务器通信的过程。
Prerequisites
Config File Locations
Configuration Structure
The config file uses a JSON object with a top-level mcpServers key. Each server is a nested object with:
command: The executable to run (e.g.,"python","npx","node")args: Array of arguments passed to the commandcwd(optional): Working directory for the server processenv(optional): Environment variables
CN: 配置文件使用一个带有顶层 mcpServers 键的 JSON 对象。每个服务器是一个嵌套对象，包含：
command: 要运行的可执行文件（如"python"、"npx"、"node"）args: 传递给命令的参数数组cwd（可选）: 服务器进程的工作目录env（可选）: 环境变量
Three Common Config Patterns
Pattern 1 — Python Server
{
"mcpServers": {
"my-python-server": {
"command": "python",
"args": ["/path/to/your/mcp_server.py"],
"cwd": "/path/to/your/project"
}
}
}Pattern 2 — npx Server
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/username/Desktop"]
}
}
}Pattern 3 — Remote Bridge
{
"mcpServers": {
"remote-server": {
"command": "npx",
"args": ["mcp-remote", "http://studio:3456/sse"]
}
}
}Complete Setup & Testing
Step 1: Verify Prerequisites
# EN: Check Node.js installation / 检查 Node.js 安装
node --version
# EN: Should output v18+ / 应输出 v18+
# EN: Check npm installation / 检查 npm 安装
npm --version
# EN: Check Python installation (if using Python server) / 检查 Python 安装（如果用 Python 服务器）
python --version
# EN: Should output Python 3.11+ / 应输出 Python 3.11+
Step 2: Create/Edit Config File (Windows Example)
Bash
C：
CD C:\Users\chene\AppData\Roaming\Claude # 如果没有建立一个
create a file, named: claude_desktop_config.json (content below)
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"C:\\Users\\yourusername\\Desktop",
"C:\\Users\\yourusername\\Downloads"
]
},
"my-python-mcp": {
"command": "python",
"args": [
"C:\\Users\\yourusername\\projects\\my-mcp-server\\mcp_server.py"
],
"cwd": "C:\\Users\\yourusername\\projects\\my-mcp-server"
}
}
}Create mcp server
according to my config file above, create a mcp_seerver.py at
"C:\\Users\\yourusername\\projects\\my-mcp-server\\mcp_server.py"
#!/usr/bin/env python3
"""
EN: Complete MCP Server with proper handshake (initialize + tools/list + tools/call).
CN: 完整的 MCP 服务器，包含正确的握手（initialize + tools/list + tools/call）。
"""
import json
import sys
def main():
"""
EN: Main loop: reads JSON-RPC lines from stdin, writes responses to stdout.
CN: 主循环：从 stdin 读取 JSON-RPC 行，向 stdout 写入响应。
"""
# EN: MCP uses stdio — keep reading line by line forever
# CN: MCP 使用 stdio —— 永远逐行持续读取
for line in sys.stdin:
# EN: Remove trailing newline and skip empty lines
# CN: 移除末尾换行符并跳过空行
line = line.strip()
if not line:
continue
try:
# EN: Parse the incoming JSON-RPC request
# CN: 解析传入的 JSON-RPC 请求
request = json.loads(line)
method = request.get("method")
req_id = request.get("id")
# ============================================================
# EN: PART 1 — MCP Handshake (MUST handle "initialize")
# CN: 第一部分 — MCP 握手（必须处理 "initialize"）
# ============================================================
if method == "initialize":
"""
EN: Respond to the initialize request with protocol version and capabilities.
Then immediately send the "initialized" notification.
This is REQUIRED for Claude Desktop to accept the connection.
CN: 用协议版本和能力响应 initialize 请求。
然后立即发送 "initialized" 通知。
这是 Claude Desktop 接受连接所必需的。
"""
# EN: Send the initialize response back to Claude Desktop
# CN: 将 initialize 响应发送回 Claude Desktop
response = {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {} # EN: We support tools / CN: 我们支持工具
},
"serverInfo": {
"name": "my-python-mcp",
"version": "1.0.0"
}
}
}
sys.stdout.write(json.dumps(response) + "\n")
sys.stdout.flush()
# EN: Send the "initialized" notification (THIS IS CRITICAL)
# CN: 发送 "initialized" 通知（这是关键）
notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
sys.stdout.write(json.dumps(notification) + "\n")
sys.stdout.flush()
# ============================================================
# EN: PART 2 — List available tools
# CN: 第二部分 — 列出可用工具
# ============================================================
elif method == "tools/list":
"""
EN: Return the list of tools this server provides.
CN: 返回此服务器提供的工具列表。
"""
response = {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"tools": [
{
"name": "hello",
"description": "Say hello to someone",
"inputSchema": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name to greet"
}
},
"required": ["name"]
}
}
]
}
}
sys.stdout.write(json.dumps(response) + "\n")
sys.stdout.flush()
# ============================================================
# EN: PART 3 — Execute a tool
# CN: 第三部分 — 执行工具
# ============================================================
elif method == "tools/call":
"""
EN: Execute the requested tool with the given arguments.
CN: 用给定的参数执行请求的工具。
"""
params = request.get("params", {})
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "hello":
name = arguments.get("name", "World")
result = {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"content": [
{
"type": "text",
"text": f"Hello, {name}! 👋"
}
]
}
}
sys.stdout.write(json.dumps(result) + "\n")
sys.stdout.flush()
except json.JSONDecodeError as e:
# EN: Handle invalid JSON gracefully
# CN: 优雅地处理无效 JSON
error_response = {
"jsonrpc": "2.0",
"id": None,
"error": {
"code": -32700,
"message": f"Parse error: {e}"
}
}
sys.stdout.write(json.dumps(error_response) + "\n")
sys.stdout.flush()
except Exception as e:
# EN: Catch any other error to prevent the server from crashing silently
# CN: 捕获任何其他错误，防止服务器静默崩溃
error_response = {
"jsonrpc": "2.0",
"id": request.get("id") if 'request' in locals() else None,
"error": {
"code": -32603,
"message": f"Internal error: {e}"
}
}
sys.stdout.write(json.dumps(error_response) + "\n")
sys.stdout.flush()
if __name__ == "__main__":
main()Step 3: Validate JSON
A trailing comma will silently break the configuration — Claude Desktop won’t show errors, tools just won’t appear.
CN: 尾随逗号会静默破坏配置——Claude Desktop 不会显示错误，工具就是出不来。
Bash
c:
cd C:\Users\chene\AppData\Roaming\Claude
# EN: Validate JSON using Python / 用 Python 验证 JSON
python -c "import json; json.load(open('claude_desktop_config.json'))"
# EN: No output means valid JSON / 无输出表示 JSON 有效
# EN: Or use online validator / 或使用在线验证器
# https://jsonlint.com/
Step 4: Restart Claude Desktop
Critical: You must fully quit Claude Desktop and reopen it. A window reload or “Refresh” is NOT enough.
CN — 关键: 你必须完全退出 Claude Desktop 再重新打开。窗口刷新或”重新加载”是不够的。
| Platform | How to fully quit | EN | CN |
|---|---|---|---|
| macOS | Cmd + Q or right-click dock icon → Quit | Command+Q 或右键 Dock 图标→退出 | Command+Q 或右键 Dock 图标→退出 |
| Windows | Right-click system tray → Exit | 右键系统托盘→退出 | 右键系统托盘→退出 |
Memory Operation Primitives
As a Senior Data Engineer, you already know CRUD operations on databases. Today, we map that familiar concept onto AI Agent memory — but with a twist: memory here is not just static data; it’s the agent’s evolving understanding of the user, the task, and the world.
作为一名资深数据工程师，你已经熟悉数据库的 CRUD 操作。今天，我们将把这个熟悉的概念映射到 AI Agent 的记忆上——但这里有一个不同之处：这里的记忆不仅仅是静态数据，而是 Agent 对用户、任务和世界不断演化的理解。
What is “Memory Operation Primitives”?
In technical terms, these are four primitivesm. They are the minimum set of operations any agent memory system must implement. They define how an agent maintains state across interactions, enabling long-running tasks, personalization, and context coherence.
Store, Recall, Update and Forget.
CN: 用技术术语说，四种原语是 任何 Agent 记忆系统必须实现的最小操作集。它们定义了 Agent 如何在多次交互中维持状态，从而实现长时间运行的任务、个性化以及上下文连贯性。
- Store = you write down a new fact the boss tells you.
- Recall = you flip through the notebook to find that fact when needed.
- Update = you cross out an old fact and write the corrected version.
- Forget = you tear out pages that are no longer relevant to save space.
CN: 想象你是一位私人助理（即 Agent），手里有一个笔记本。
- Store（保存） = 你把老板告诉你的新事实写下来。
- Recall（回忆/检索） = 你需要时翻阅笔记本找到那条事实。
- Update（更新） = 你划掉旧事实并写上修正后的版本。
- Forget（遗忘/删除） = 你把不再相关的页撕掉，以节省空间。
Without these four operations, your agent is stateless — every conversation starts from zero. That’s fine for a simple Q&A bot, but useless for a real AI/LLM Engineer’s agent that needs to:
- Remember user preferences across sessions (e.g., “always use Databricks SQL warehouse”).
- Recall previous task results to avoid recomputation.
- Update its understanding as new data arrives.
- Forget outdated information to prevent hallucination and context overflow.
CN: 没有这四个操作，你的 Agent 就是无状态的——每次对话都从零开始。这对于简单的问答机器人还行，但对于一个真正的 AI/LLM 工程师的 Agent 来说毫无用处，因为它需要：
- 跨会话记住用户偏好（例如“始终使用 Databricks SQL warehouse”）。
- 回忆先前任务结果以避免重复计算。
- 随着新数据到达而更新其理解。
- 遗忘过时信息以防止幻觉和上下文溢出。
More importantly, these primitives are the building blocks for Short-term Memory (conversation state), Long-term Memory (user preferences), Episodic Memory (historical tasks), and Semantic Memory (knowledge base) — all of which appear later in your plan.
CN: 更重要的是，这些原语是短期记忆（对话状态）、长期记忆（用户偏好）、情景记忆（历史任务）和语义记忆（知识库）的构建块——这些都会在你后续的学习计划中出现。
What does each primitive contain?
1. Store (保存)
| Aspect | EN | CN |
|---|---|---|
| Definition | Persist a new memory item into the memory system. | 将一条新的记忆条目持久化到记忆系统中。 |
| Input | Key (or ID), Content (text/structured data), Metadata (timestamp, type, importance), optionally a TTL (time-to-live). | Key（或ID）、内容（文本/结构化数据）、元数据（时间戳、类型、重要性）、可选 TTL（生存时间）。 |
| Output | Confirmation of storage, or the generated ID. | 存储确认，或生成的ID。 |
| Example | Store: user_preference → "preferred_cluster = 'large'" | 保存：user_preference → "preferred_cluster = 'large'" |
2. Recall (回忆/检索)
| Aspect | EN | CN |
|---|---|---|
| Definition | Retrieve a memory item by key, or perform a query (semantic/keyword) to find relevant memories. | 通过 key 检索一条记忆条目，或执行查询（语义/关键词）以找到相关记忆。 |
| Input | Key (exact lookup) or Query string + optional filters (type, date range). | Key（精确查找）或查询字符串 + 可选过滤器（类型、日期范围）。 |
| Output | The memory item(s) or None. | 记忆条目或 None。 |
| Example | Recall user_preference → returns "preferred_cluster = 'large'" | 回忆 user_preference → 返回 "preferred_cluster = 'large'" |
3. Update (更新)
| Aspect | EN | CN |
|---|---|---|
| Definition | Modify an existing memory item — either full replacement or partial merge. | 修改已有的记忆条目——可以是完全替换或部分合并。 |
| Input | Key, new content, merge strategy (overwrite vs. merge). | Key、新内容、合并策略（覆盖 vs. 合并）。 |
| Output | Updated item or status. | 更新后的条目或状态。 |
| Example | Update user_preference → "preferred_cluster = 'medium'" | 更新 user_preference → "preferred_cluster = 'medium'" |
4. Forget (遗忘/删除)
| Aspect | EN | CN |
|---|---|---|
| Definition | Remove a memory item, either hard-delete or soft-delete (mark expired). | 移除一条记忆条目，可以是硬删除或软删除（标记过期）。 |
| Input | Key or a condition (e.g., forget all memories older than 7 days). | Key 或一个条件（例如，遗忘所有超过7天的记忆）。 |
| Output | Status (success/failure). | 状态（成功/失败）。 |
| Example | Forget user_preference → removes it. | 遗忘 user_preference → 移除它。 |
Key Takeaways
| 要点 (Aspect) | EN (English) | CN (中文) |
|---|---|---|
| Four Primitives | Store, Recall, Update, Forget are the minimum CRUD operations for agent memory. | Store、Recall、Update、Forget 是 Agent 记忆的最小 CRUD 操作集。 |
| Store | Persists a new memory with optional TTL and metadata. | 持久化一条新记忆，可带 TTL 和元数据。 |
| Recall | Retrieves by key, automatically checking expiration. | 按 key 检索，自动检查过期。 |
| Update | Supports both full replacement and partial merge (for dict content). | 支持完全替换和部分合并（针对 dict 内容）。 |
| Forget | Hard-delete or soft-delete; we implemented hard-delete. | 硬删除或软删除；我们实现了硬删除。 |
| TTL (Time-to-Live) | Expired items are automatically forgotten on recall, preventing context bloat. | 过期的条目在 recall 时自动遗忘，防止上下文膨胀。 |
| Backend Abstraction | Separate interface from implementation for easy swapping (in-memory vs SQLite). | 将接口与实现分离，便于切换（内存 vs SQLite）。 |
| Metadata | Always attach metadata (type, importance, source) to enable advanced filtering later. | 始终附加元数据（类型、重要性、来源），以便以后实现高级过滤。 |
| Async Consideration | For production, use async database drivers (e.g., aiosqlite, asyncpg). We’ll cover async in A17. | 生产环境中使用异步数据库驱动（如 aiosqlite, asyncpg）。我们将在 A17 中讨论异步。 |
| Why It Matters | These primitives form the foundation for all four memory types (Short, Long, Episodic, Semantic) in your plan. | 这些原语构成了你计划中所有四种记忆类型（短期、长期、情景、语义）的基础。 |
Simple RAG Pipeline
RAG stands for Retrieval-Augmented Generation. It’s a technique that combines information retrieval with LLM generation — instead of asking the LLM to answer from memory alone, you first retrieve relevant documents from an external knowledge base, then feed those documents as context to the LLM, and let the LLM generate an answer based on that context.
Why It Matters?
Traditional LLMs have four fatal flaws that RAG solves:
Simple RAG Pipeline
A simple RAG pipeline has two phases and five stages:
- Phase 1: Indexing (Offline / 离线阶段)
Build the searchable knowledge base before users ask questions.
离线阶段（索引）：先把所有文档加载进来，切分成小块，调用 Embedding 模型转成向量，最后存进 FAISS 等向量数据库里。 - Phase 2: Retrieval & Generation (Online / 在线阶段)
Execute at query time for each user question.
在线阶段（检索+生成）：用户提问时，先把问题也转成向量，去库里找最相似的 Top-K 个块，把这几个块作为“参考资料”连同问题一起扔给 GPT，让它写出最终回答。
Stage-by-Stage Breakdown
Code Implementation
Below is a complete, self-contained simple RAG pipeline. We’ll use:
- LangChain for orchestration
- FAISS as vector database (local, no server needed)
- OpenAI for embeddings and chat completion
Runtime Sequence
#!/usr/bin/env python3
"""
================================================================================
B02 升级版：完整集成 (Code A + Code B + 智能缓存)
================================================================================
这是一个 100% 完整的、可直接复制粘贴运行的 Python 脚本。
包含：
1. 数据加载 (Load)
2. 文本切分 (Split)
3. FAISS 专业配置 (支持 Flat / IVF 索引，带持久化)
4. RAG 检索生成链 (Retrieve + Generate)
5. 智能缓存逻辑 (首次构建，后续秒级加载)
使用方法：
1. 安装依赖：pip install faiss-cpu langchain langchain-community langchain-openai openai numpy
2. 设置环境变量：export OPENAI_API_KEY="你的Key"
3. 运行：python rag_complete.py
================================================================================
"""
# ============================================================
# 0. 依赖导入 (全部列出，一个不少)
# ============================================================
import os
import tempfile
import shutil
from typing import List, Dict, Any, Optional
import numpy as np
import faiss
# LangChain 核心
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
# LangChain 社区
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores.utils import DistanceStrategy
# LangChain 集成
from langchain_openai import ChatOpenAI
# 文本切分
from langchain_text_splitters import RecursiveCharacterTextSplitter
# ============================================================
# 第一部分：CODE A —— 原版 RAG 流程 (加载、切分、生成链)
# ============================================================
def load_documents(file_paths: List[str]) -> List[Document]:
"""
从文本文件加载文档。
"""
all_docs: List[Document] = []
for path in file_paths:
loader = TextLoader(path, encoding="utf-8")
docs = loader.load()
all_docs.extend(docs)
print(f"✅ 加载完成: {path} ({len(docs)} 个文档)")
return all_docs
def chunk_documents(documents: List[Document]) -> List[Document]:
"""
将大文档切分成语义块。
使用 RecursiveCharacterTextSplitter 保证上下文连贯。
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块最大字符数
chunk_overlap=200, # 块间重叠，保留跨块上下文
length_function=len,
separators=["\n\n", "\n", " ", ""],
add_start_index=True, # 标记在原文中的位置
)
chunks = text_splitter.split_documents(documents)
print(f"✅ 切分完成: {len(documents)} 个文档 → {len(chunks)} 个块")
return chunks
def retrieve_context(vector_store: FAISS, query: str, k: int = 4) -> List[Document]:
"""
从向量库中检索最相似的 k 个块。
"""
retrieved_docs = vector_store.similarity_search(query, k=k)
print(f"🔍 检索到 {len(retrieved_docs)} 个块")
return retrieved_docs
def format_context(documents: List[Document]) -> str:
"""
将检索到的文档格式化成上下文字符串。
"""
return "\n\n---\n\n".join([
f"[来源 {i+1}]\n{doc.page_content}"
for i, doc in enumerate(documents)
])
def create_rag_chain(vector_store: FAISS):
"""
使用 LCEL 构建完整的 RAG 生成链。
这是 Code A 的核心生成逻辑，完全不需要改动。
"""
# 提示词模板：强制 LLM 仅基于上下文回答
prompt_template = ChatPromptTemplate.from_template("""
你是一个只根据提供的上下文来回答问题的助手。
如果上下文中没有答案，请直接说"我没有足够的信息回答这个问题"。
上下文 (Context):
{context}
问题 (Question):
{question}
答案 (Answer):
""")
# 使用 gpt-4o-mini，temperature=0 保证事实性回答
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
# 内部函数：检索 + 格式化
def retrieve_and_format(inputs: Dict[str, Any]) -> Dict[str, str]:
question = inputs["question"]
docs = retrieve_context(vector_store, question)
context = format_context(docs)
return {"context": context, "question": question}
# 构建 LCEL 链
rag_chain = (
RunnablePassthrough()
| RunnableLambda(retrieve_and_format)
| prompt_template
| llm
| StrOutputParser()
)
return rag_chain
# ============================================================
# 第二部分：CODE B —— FAISS 专业配置工厂 (带索引工程 + 持久化)
# ============================================================
class FAISSAdvancedConfig:
"""
FAISS 高级配置类。
控制距离度量、索引类型、检索参数。
"""
def __init__(
self,
index_factory_string: str = "Flat",
nprobe: int = 10
):
self.distance_strategy = DistanceStrategy.COSINE # 余弦相似度
self.index_factory_string = index_factory_string # 例如 "Flat", "IVF100,Flat"
self.nprobe = nprobe # IVF 查询时探测的簇数
class FAISSFactory:
"""
FAISS 工厂类。
职责：构建索引、保存到磁盘、从磁盘加载。
"""
def __init__(self, config: FAISSAdvancedConfig):
self.config = config
# 嵌入模型：text-embedding-3-small (1536维，性价比高)
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
def build_from_documents(self, chunks: List[Document]) -> FAISS:
"""
从文档块构建 FAISS 索引。
支持 Flat / IVF / HNSW 等所有 FAISS 原生索引类型。
"""
print(f"🚀 正在构建 FAISS 索引 (类型: {self.config.index_factory_string})...")
# 1. 提取文本并生成向量
texts = [doc.page_content for doc in chunks]
vectors = self.embeddings.embed_documents(texts) # 返回 List[List[float]]
vectors_np = np.array(vectors).astype('float32') # FAISS 要求 float32
dim = vectors_np.shape[1] # 向量维度 (1536)
n_vectors = vectors_np.shape[0]
print(f" 向量维度: {dim}, 向量数量: {n_vectors}")
# 2. 使用工厂字符串创建原生 FAISS 索引
# "Flat" -> 暴力搜索 (精确)
# "IVF100,Flat" -> 100个聚类的倒排索引 (快速)
index = faiss.index_factory(dim, self.config.index_factory_string)
# 3. 如果使用 IVF，需要训练 (聚类) 并设置 nprobe
if "IVF" in self.config.index_factory_string:
print(f" 正在训练 IVF (nlist={self.config.index_factory_string.split('IVF')[1].split(',')[0]})...")
index.train(vectors_np)
# nprobe 越大，召回率越高，速度越慢
index.nprobe = self.config.nprobe
print(f" 设置 nprobe = {self.config.nprobe}")
# 4. 添加向量到索引
index.add(vectors_np)
print(f" 成功添加 {n_vectors} 个向量")
# 5. 包装成 LangChain 的 FAISS 对象
# 需要构建 docstore (存储原始文本) 和 id 映射
docstore = {str(i): chunk for i, chunk in enumerate(chunks)}
index_to_docstore_id = {i: str(i) for i in range(len(chunks))}
vector_store = FAISS(
embedding_function=self.embeddings.embed_query, # 查询时用的函数
index=index,
docstore=docstore,
index_to_docstore_id=index_to_docstore_id,
relevance_score_fn=None # LangChain 自动根据 distance_strategy 处理
)
print("✅ FAISS 索引构建完成")
return vector_store
def save(self, vector_store: FAISS, path: str) -> None:
"""
将 FAISS 索引保存到磁盘。
会生成两个文件: {path}.pkl 和 {path}.faiss
"""
vector_store.save_local(path)
print(f"💾 索引已保存到: {path}")
def load(self, path: str) -> FAISS:
"""
从磁盘加载 FAISS 索引。
"""
print(f"📂 正在从磁盘加载索引: {path}")
# allow_dangerous_deserialization=True 是因为本地开发环境安全
vector_store = FAISS.load_local(
folder_path=path,
embeddings=self.embeddings,
allow_dangerous_deserialization=True
)
print("✅ 索引加载成功")
return vector_store
# ============================================================
# 第三部分：集成逻辑 (智能缓存 + 自动切换)
# ============================================================
def get_vector_store_with_cache(
chunks: List[Document],
cache_path: str = "./faiss_cache",
force_rebuild: bool = False
) -> FAISS:
"""
智能获取向量存储：
- 如果 cache_path 存在且 force_rebuild=False -> 直接加载 (秒级启动)
- 否则 -> 用专业配置构建，并保存到 cache_path
自动索引选型：
- 向量数 <= 10000 -> 使用 Flat (暴力精确)
- 向量数 > 10000 -> 使用 IVF (倒排加速)
"""
# 根据数据量自动选择索引类型
if len(chunks) <= 10000:
index_type = "Flat"
else:
index_type = "IVF100,Flat" # 100个聚类，适合中等规模
print(f"⚙️ 数据量: {len(chunks)} 条，选用索引: {index_type}")
# 创建配置和工厂
config = FAISSAdvancedConfig(index_factory_string=index_type, nprobe=10)
factory = FAISSFactory(config)
# 决策：加载缓存 or 重新构建
if not force_rebuild and os.path.exists(cache_path):
# 路径存在 -> 直接加载
return factory.load(cache_path)
else:
# 路径不存在 或 强制重建 -> 构建并保存
if force_rebuild:
print("🔄 强制重建索引...")
vector_store = factory.build_from_documents(chunks)
factory.save(vector_store, cache_path)
return vector_store
# ============================================================
# 第四部分：主程序入口 (包含完整测试数据)
# ============================================================
def main():
"""
完整流程演示：
1. 生成临时测试文档
2. 加载 + 切分
3. 获取向量存储 (带缓存)
4. 执行 RAG 问答
5. 清理临时文件
"""
print("=" * 70)
print("B02 升级版 RAG Pipeline (Code A + Code B 完整集成)")
print("=" * 70)
# ----- 1. 准备测试数据 (临时文件) -----
print("\n📝 创建临时测试文档...")
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".txt",
delete=False,
encoding="utf-8"
) as f:
f.write("""
Retrieval-Augmented Generation (RAG) is a technique that combines
information retrieval with large language model generation.
RAG has two main phases: indexing and retrieval-generation.
The indexing phase involves loading documents, splitting them into chunks,
converting chunks to embeddings, and storing them in a vector database.
The retrieval-generation phase involves converting a user query to an
embedding, searching the vector database for similar chunks, and using
those chunks as context for the LLM to generate an answer.
RAG solves the problem of LLM hallucinations by grounding answers in
retrieved facts. It also allows LLMs to access private or up-to-date
information that wasn't in their training data.
FAISS (Facebook AI Similarity Search) is a library for efficient
similarity search and clustering of dense vectors. It is widely used
as the vector database in RAG systems.
""")
temp_path = f.name
# ----- 2. 加载 + 切分 -----
print("\n📂 阶段 1: 加载与切分")
print("-" * 40)
docs = load_documents([temp_path])
chunks = chunk_documents(docs)
# ----- 3. 获取向量存储 (智能缓存) -----
print("\n💾 阶段 2: 向量存储初始化")
print("-" * 40)
# 首次运行会构建，第二次运行会直接加载 (秒级)
cache_dir = "./demo_faiss_cache"
vector_store = get_vector_store_with_cache(
chunks=chunks,
cache_path=cache_dir,
force_rebuild=False # 设为 True 可强制重建
)
# ----- 4. 执行 RAG 问答 -----
print("\n🤖 阶段 3: RAG 问答")
print("-" * 40)
# 创建 RAG 链 (完全使用 Code A 的逻辑)
rag_chain = create_rag_chain(vector_store)
# 测试问题列表
questions = [
"What is RAG?",
"What are the two main phases of RAG?",
"How does RAG solve the hallucination problem?",
"What is FAISS?"
]
for i, question in enumerate(questions, 1):
print(f"\nQ{i}: {question}")
answer = rag_chain.invoke({"question": question})
print(f"A{i}: {answer}")
# ----- 5. 清理临时文件 -----
print("\n🧹 清理临时文件...")
os.unlink(temp_path) # 删除临时 txt
# 保留缓存目录，方便下次测试秒启；如果想删掉，取消注释下面一行
# shutil.rmtree(cache_dir, ignore_errors=True)
print("\n" + "=" * 70)
print("✅ RAG Pipeline 运行完毕")
print(f"💡 缓存目录: {cache_dir} (保留以加速下次启动)")
print("=" * 70)
if __name__ == "__main__":
main()逐阶段详解
阶段 0：环境准备 (Pre-flight)
| 步骤 | 做什么 | 为什么重要 |
|---|---|---|
| 0.1 | pip install faiss-cpu langchain... | 确保运行环境就绪 |
| 0.2 | export OPENAI_API_KEY="sk-..." | LLM 和 Embedding 服务需要鉴权 |
| 0.3 | 导入 os, numpy, faiss, langchain 全套 | 代码依赖的基础设施 |
阶段 1：数据加载 (Data Ingestion)
| 步骤 | 做什么 | 对应代码函数 | 输入 → 输出 |
|---|---|---|---|
| 1.1 | 创建临时 .txt 文件（或指定真实文件路径） | tempfile.NamedTemporaryFile | 文本字符串 → 物理文件 |
| 1.2 | 读取文件内容，包装成 LangChain Document 对象 | load_documents() | 文件路径 → List[Document] |
为什么：
Document是 LangChain 的标准数据单元，包含page_content（文本）和metadata（元数据）。
阶段 2：文本切分 (Chunking)
| 步骤 | 做什么 | 对应代码函数 | 关键参数 |
|---|---|---|---|
| 2.1 | 初始化递归切分器 | RecursiveCharacterTextSplitter | chunk_size=1000, overlap=200 |
| 2.2 | 执行切分 | split_documents() | 长文档 → 多个短块 |
为什么：
- 防止 Embedding 模型截断（通常最大输入 8192 tokens，但 1000 字符更稳妥）。
- 200 字符重叠保证跨段落的语义连贯（比如一句话被切到两段时，重叠部分能保留完整语义）。
阶段 3：智能缓存路由 (Cache Router) —— 这是集成代码 B 的核心价值
| 步骤 | 做什么 | 决策逻辑 |
|---|---|---|
| 3.1 | 检查磁盘目录 ./faiss_cache 是否存在 | os.path.exists(cache_path) |
| 3.2 | 分支 A (首次运行)：目录不存在 → 执行阶段 4A | 构建全新的索引 |
| 3.2 | 分支 B (后续运行)：目录存在 → 执行阶段 4B | 直接加载，跳过昂贵的 Embedding 计算 |
阶段 4A：索引构建 (Build) —— 离线阶段，耗时但只做一次
| 子步骤 | 做什么 | 对应代码 |
|---|---|---|
| 4A.1 | 调用 OpenAI Embedding API，把每个 chunk 变成 1536 维向量 | embeddings.embed_documents(texts) |
| 4A.2 | 根据数据量自动选型：≤1万用 Flat（精确），>1万用 IVF100,Flat（加速） | faiss.index_factory(dim, index_type) |
| 4A.3 | 如果是 IVF，执行 K-Means 聚类训练 (index.train) | 把向量划分到 100 个簇里 |
| 4A.4 | 将所有向量添加到索引 (index.add) | 构建倒排表 |
| 4A.5 | 将索引和原始文本打包成 LangChain FAISS 对象 | FAISS(...) 构造函数 |
| 4A.6 | 保存到磁盘：./faiss_cache.pkl + ./faiss_cache.faiss | factory.save() |
阶段 4B：索引加载 (Load) —— 在线阶段，毫秒级恢复
| 子步骤 | 做什么 | 对应代码 |
|---|---|---|
| 4B.1 | 读取磁盘上的 .pkl 文件（包含 docstore 和元数据） | pickle.load() |
| 4B.2 | 读取 .faiss 二进制文件（包含索引结构） | faiss.read_index() |
| 4B.3 | 重建内存中的 LangChain FAISS 对象 | 直接返回，无需调用任何 API |
为什么这很重要：如果不做持久化，每次重启服务，阶段 4A 都要重新调用 OpenAI API 嵌入所有文档，对于 10 万条数据，耗费数小时且产生巨额费用。
阶段 5：构建 RAG 链 (Chain Assembly)
| 步骤 | 做什么 | 对应代码 | 本质 |
|---|---|---|---|
| 5.1 | 定义 System Prompt（约束 LLM 只能基于上下文回答） | ChatPromptTemplate.from_template | 提示词工程 |
| 5.2 | 将“检索+格式化”封装成可调用函数 | RunnableLambda(retrieve_and_format) | 定制化检索逻辑 |
| 5.3 | 用 LCEL 的 | 运算符串联所有组件 | ... | prompt | llm | parser | 构建流水线（Pipeline） |
| 5.4 | 编译成可执行的 Runnable 对象 | rag_chain | 等待输入 |
阶段 6：问答执行 (Inference) —— 在线阶段，每次查询触发
| 子步骤 | 做什么 | 具体动作 | 耗时占比 |
|---|---|---|---|
| 6.1 | 检索 (Retrieve) | vector_store.similarity_search(query, k=4) → 返回 4 个最相似的 chunks | ~10% |
| 6.2 | 增强 (Augment) | 把 4 个 chunks 拼接成一个长字符串（Context） | ~1% |
| 6.3 | 生成 (Generate) | 调用 gpt-4o-mini API，传入 Context + Question | ~89% (网络IO) |
| 6.4 | 解析 (Parse) | 提取 LLM 返回的文本内容 | <1% |
阶段 7：资源回收 (Cleanup)
| 步骤 | 做什么 | 原因 |
|---|---|---|
| 7.1 | os.unlink(temp_path) | 删除测试用的临时 .txt 文件，保持工作区整洁 |
| 7.2 | (可选) shutil.rmtree(cache_path) | 如果想完全重置，删除缓存目录；否则保留加速下次启动 |
How to run?
# 1. 安装依赖（如果还没装）
pip install faiss-cpu langchain langchain-community langchain-openai openai numpy
# 2. 设置 OpenAI API Key
export OPENAI_API_KEY="sk-你的Key"
# 3. 把上面的完整代码保存为 rag_complete.py，然后运行
python rag_complete.py
第一次运行：会构建索引并保存到 ./demo_faiss_cache 目录（大约需要 3-5 秒）。
第二次运行：会直接加载缓存（不到 1 秒就启动），你可以把 force_rebuild=False 改成 True 来测试强制重建。Key Takeaways
速查表：逻辑流程 vs 代码行号对照
| 逻辑阶段 | 主函数调用 | 代码文件中的位置 |
|---|---|---|
| 0. 环境准备 | import ... + OPENAI_API_KEY | 顶部 1-40 行 |
| 1. 数据加载 | load_documents([temp_path]) | main() 函数内部 |
| 2. 文本切分 | chunk_documents(docs) | main() 函数内部 |
| 3. 缓存路由 | get_vector_store_with_cache(chunks, cache_path) | main() 函数内部 |
| 4A. 构建索引 | factory.build_from_documents(chunks) | FAISSFactory 类内部 |
| 4B. 加载索引 | factory.load(cache_path) | FAISSFactory 类内部 |
| 5. 构建链 | create_rag_chain(vector_store) | main() 函数内部 |
| 6. 问答循环 | rag_chain.invoke({"question": q}) | for 循环内部 |
| 7. 资源回收 | os.unlink(temp_path) | main() 函数末尾 |
总结一句话
这个流程本质上是一个 ETL Pipeline 的 AI 变体：
| 传统 ETL | 这个 RAG 流程 |
|---|---|
| Extract (抽取) | 阶段 1：加载文档 |
| Transform (转换) | 阶段 2：切分文本 + 阶段 4A：向量化 |
| Load (加载) | 阶段 4A/4B：存入/读取 FAISS 向量库 |
| 查询 (Query) | 阶段 6：检索 + LLM 生成 |
现在你手里既有完整的代码，又有清晰的流程图，完全可以按图索骥，把每一块代码对应到具体的业务步骤上。你可以把这个流程图画在你的工程文档里，作为系统设计的一部分。
Embedding
Embedding is a fundamental technology in machine learning and natural language processing that transforms discrete or complex objects (such as words, sentences, or images) into numerical vector representations of a fixed dimension.
In a RAG (Retrieval-Augmented Generation) system, embedding is the foundation of retrieval. Without embeddings, you cannot perform semantic search — you would be stuck with keyword matching only
Core Concepts
Vector Space & Similarity
Embeddings live in a multi-dimensional vector space. To measure how similar two pieces of text are, we measure the distance between their embedding vectors. The most common measure is cosine similarity.
存在于一个多维向量空间中。要衡量两段文本有多相似，我们衡量它们的 Embedding 向量之间的距离。最常用的度量是余弦相似度（Cosine Similarity）.
Cosine Similarity explained:
- Range: -1 (opposite) to 1 (identical)
-1（相反）到 1（完全相同） - For text embeddings, values close to 1 mean high semantic similarity
对于文本 Embedding，接近 1 的值表示高语义相似度 - It measures the cosine of the angle between two vectors, ignoring magnitude
它测量两个向量之间的夹角余弦值，忽略向量大小
import numpy as np
def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
"""
Calculate cosine similarity between two vectors.
计算两个向量之间的余弦相似度。
Purpose: Measures semantic similarity between two embeddings.
目的：衡量两个 Embedding 之间的语义相似度。
"""
# Normalize vectors / 归一化向量
norm_a = np.linalg.norm(vec_a)
norm_b = np.linalg.norm(vec_b)
# Dot product / 点积
dot_product = np.dot(vec_a, vec_b)
# Cosine similarity = dot product / (norm_a * norm_b)
# 余弦相似度 = 点积 / (norm_a * norm_b)
return dot_product / (norm_a * norm_b)Dense vs Sparse Vectors
稠密向量 vs 稀疏向量
| Aspect | Dense Embedding | Sparse (One-Hot) |
|---|---|---|
| Dimension | Fixed, low (e.g., 384, 768, 1536) | Very high (vocabulary size) |
| Values | Continuous floats | 0 or 1 |
| Semantic info 语义信息 | Captures meaning 捕获含义 | Only captures identity 仅捕获身份 |
| Efficiency | Computationally efficient | Wastes memory |
Sparse (One-Hot) 中，每个词是一个独立的维度，”king”是 [0,0,0,…,1,…,0]，”queen”是另一个位置。它们没有任何语义关系。而 Embedding 中，”king”和”queen”的向量是相似的。
Captures meaning 捕获含义 = 不看字，看“意思像不像”
Only captures identity 仅捕获身份 = 只认“是不是这个词”，不管什么意思
Contextual vs Static Embeddings
上下文 Embedding vs 静态 Embedding
| Type | Examples | Characteristics |
|---|---|---|
| Static | Word2Vec, GloVe, FastText | One vector per word, regardless of context 每个词一个向量，无论上下文 |
| Contextual | BERT-based, OpenAI text-embedding-3 | Vector changes based on surrounding words 向量根据周围词变化 |
- Word2Vec: Word2Vec is a method that learns word meanings by looking at the context in which words appear.
是一种通过“上下文”来学习词语含义的模型。它通过“上下文预测”学习词向量的方法。看词和词之间的搭配关系来学习语义,让语义相似的词在向量空间中靠得更近。 - GloVe = Global Vectors for Word Representation. It learns word meaning from: global word co-occurrence statistics
“统计全世界词和词的关系” - FastText：FastText (Facebook) represents words as： sum of character n-grams把“词拆成字母/子词”来学向量
“bank” in “river bank” vs “bank account” ： static embedding gives the same vector; contextual embedding gives different vectors based on context.
静态 Embedding 给相同的向量；上下文 Embedding 根据上下文给不同的向量。
BERT = Bidirectional Encoder Representations from Transformers
It learns: context-aware word meaning using Transformer attention
用 Transformer（注意力机制）来理解上下文的语言模型。
“河岸”，“银行账号” “同一个词，在不同句子里有不同含义”
Mainstream Embedding Models
For modern RAG systems, the most relevant embedding models are
| Model | Provider | Dim | Notes |
|---|---|---|---|
text-embedding-3-small | OpenAI | 1536 | Cost-effective, good for most RAG |
text-embedding-3-large | OpenAI | 3072 | Higher quality, higher cost |
text-embedding-ada-002 | OpenAI | 1536 | Legacy, being replaced by v3 |
BGE-M3 | BAAI | 1024 | Open-source, multilingual |
KaLM-Embedding-Gemma3-12B | Tencent | 3840 | SOTA on MMTEB |
Code Implementation
OpenAI API generates Embedding：
# ============================================================
# 1. IMPORTS / 导入依赖
# ============================================================
import os
import numpy as np
from openai import OpenAI
from typing import List, Dict, Any
# ============================================================
# 2. CLIENT INITIALIZATION / 客户端初始化
# ============================================================
# Purpose: Initialize the OpenAI client with API key from environment.
# 目的：使用环境变量中的 API Key 初始化 OpenAI 客户端。
client = OpenAI(
api_key=os.environ.get("OPENAI_API_KEY"), # Get from env / 从环境变量获取
)
# ============================================================
# 3. EMBEDDING FUNCTION / Embedding 生成函数
# ============================================================
def get_embedding(
text: str,
model: str = "text-embedding-3-small"
) -> List[float]:
"""
Generate an embedding vector for a given text using OpenAI API.
使用 OpenAI API 为给定文本生成 Embedding 向量。
Purpose: Convert text into a numerical vector for semantic search.
目的：将文本转换为数值向量，用于语义搜索。
Args:
text: Input text string / 输入文本字符串
model: Embedding model name / Embedding 模型名称
Returns:
List[float]: Embedding vector of length model-specific dimension
List[float]: Embedding 向量，长度为模型指定的维度
Important: The same model must be used for both query and document embeddings
重点：查询和文档的 Embedding 必须使用同一个模型[reference:30]
"""
# Remove newlines and excess whitespace for cleaner embedding
# 移除换行和多余空白，获得更干净的 Embedding
text = text.replace("\n", " ")
# Call OpenAI Embeddings API / 调用 OpenAI Embeddings API
# Purpose: Send text to OpenAI and get back the embedding vector
# 目的：发送文本到 OpenAI，获取 Embedding 向量
response = client.embeddings.create(
model=model, # Which embedding model to use / 使用的模型
input=text, # Text to embed / 要嵌入的文本
encoding_format="float" # Return as list of floats / 以浮点数列表返回
)
# Extract the embedding from response / 从响应中提取 Embedding
# Purpose: The embedding is in response.data[0].embedding
# 目的：Embedding 位于 response.data[0].embedding 中
embedding = response.data[0].embedding
return embedding
# ============================================================
# 4. BATCH EMBEDDING / 批量 Embedding
# ============================================================
def get_embeddings_batch(
texts: List[str],
model: str = "text-embedding-3-small"
) -> List[List[float]]:
"""
Generate embeddings for multiple texts in a single API call.
在单个 API 调用中为多个文本生成 Embedding。
Purpose: More efficient than calling get_embedding() in a loop.
目的：比在循环中调用 get_embedding() 更高效。
Important: OpenAI limits total tokens to 300,000 per request[reference:31]
重点：OpenAI 限制每个请求总 token 不超过 300,000[reference:32]
"""
# Clean texts / 清理文本
cleaned_texts = [t.replace("\n", " ") for t in texts]
# Batch API call / 批量 API 调用
response = client.embeddings.create(
model=model,
input=cleaned_texts, # List of texts / 文本列表
encoding_format="float"
)
# Extract all embeddings / 提取所有 Embedding
# Purpose: Each response.data[i] corresponds to texts[i]
# 目的：每个 response.data[i] 对应 texts[i]
embeddings = [item.embedding for item in response.data]
return embeddings
# ============================================================
# 5. SIMILARITY SEARCH / 相似度搜索
# ============================================================
def cosine_similarity(vec_a: List[float], vec_b: List[float]) -> float:
"""
Calculate cosine similarity between two embedding vectors.
计算两个 Embedding 向量之间的余弦相似度。
Purpose: Measure semantic similarity between query and document.
目的：衡量查询和文档之间的语义相似度。
Important: Embeddings must be from the SAME model.
重点：Embeddings 必须来自同一个模型。
"""
# Convert to numpy arrays for efficient math / 转换为 numpy 数组以便高效计算
a = np.array(vec_a)
b = np.array(vec_b)
# Calculate cosine similarity / 计算余弦相似度
# Formula: cos(θ) = (A · B) / (||A|| * ||B||)
# 公式：cos(θ) = (A · B) / (||A|| * ||B||)
dot_product = np.dot(a, b)
norm_a = np.linalg.norm(a)
norm_b = np.linalg.norm(b)
# Avoid division by zero / 避免除以零
if norm_a == 0 or norm_b == 0:
return 0.0
return float(dot_product / (norm_a * norm_b))
def find_most_similar(
query: str,
documents: List[str],
model: str = "text-embedding-3-small",
top_k: int = 3
) -> List[Dict[str, Any]]:
"""
Find the most semantically similar documents to a query.
找到与查询语义最相似的文档。
Purpose: Core retrieval function for RAG systems.
目的：RAG 系统的核心检索函数[reference:33]。
Args:
query: User question / 用户问题
documents: List of document chunks / 文档块列表
model: Embedding model / Embedding 模型
top_k: Number of top results to return / 返回前 K 个结果
Returns:
List of dicts with document text and similarity score
包含文档文本和相似度分数的字典列表
"""
# Step 1: Embed the query / 第一步：嵌入查询
# Purpose: Convert user question to vector for comparison
# 目的：将用户问题转换为向量以便比较
query_embedding = get_embedding(query, model)
# Step 2: Embed all documents / 第二步：嵌入所有文档
# Purpose: Convert all documents to vectors
# 目的：将所有文档转换为向量
doc_embeddings = get_embeddings_batch(documents, model)
# Step 3: Calculate similarities / 第三步：计算相似度
# Purpose: Compare query vector against all document vectors
# 目的：将查询向量与所有文档向量比较
results = []
for i, doc_embedding in enumerate(doc_embeddings):
score = cosine_similarity(query_embedding, doc_embedding)
results.append({
"document": documents[i],
"score": score,
"index": i
})
# Step 4: Sort by score descending and return top_k
# 第四步：按分数降序排序，返回 top_k
# Purpose: Return the most relevant documents first
# 目的：首先返回最相关的文档
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
# ============================================================
# 6. USAGE EXAMPLE / 使用示例
# ============================================================
if __name__ == "__main__":
# Example: RAG retrieval scenario / 示例：RAG 检索场景
# User question / 用户问题
query = "What is the capital of France?"
# Document chunks from a knowledge base / 来自知识库的文档块
documents = [
"France is a country in Western Europe. Its capital is Paris.",
"Germany's capital is Berlin. It is the largest city in Germany.",
"The Eiffel Tower is located in Paris, France.",
"London is the capital of the United Kingdom."
]
# Find most relevant documents / 找到最相关的文档
top_results = find_most_similar(query, documents, top_k=2)
print(f"Query: {query}")
print("\nTop Results:")
for i, result in enumerate(top_results):
print(f"{i+1}. Score: {result['score']:.4f}")
print(f" Document: {result['document']}")
print()
# Expected output: Documents about France and Paris should rank highest
# 预期输出：关于法国和巴黎的文档应该排名最高Azure OpenAI generates Embedding（Azure AI Foundry intgerate）
# ============================================================
# AZURE OPENAI EMBEDDING / Azure OpenAI Embedding
# ============================================================
from openai import AzureOpenAI
# Initialize Azure OpenAI client / 初始化 Azure OpenAI 客户端
# Purpose: Use Azure OpenAI Service instead of OpenAI's direct API
# 目的：使用 Azure OpenAI 服务替代 OpenAI 的直接 API
azure_client = AzureOpenAI(
api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
api_version="2024-02-15-preview", # Azure API version / Azure API 版本
azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
)
def get_azure_embedding(text: str, deployment: str = "text-embedding-3-small") -> List[float]:
"""
Generate embedding using Azure OpenAI Service.
使用 Azure OpenAI 服务生成 Embedding。
Purpose: Enterprise-grade embedding with Azure's infrastructure.
目的：使用 Azure 基础设施的企业级 Embedding。
"""
response = azure_client.embeddings.create(
model=deployment, # Deployment name in Azure / Azure 中的部署名称
input=text,
encoding_format="float"
)
return response.data[0].embeddingKey Takeaways
| 要点 | EN | CN |
|---|---|---|
| Embedding 是将文本转换为数值向量 | Embedding converts text to numerical vectors | Embedding 将文本转换为数值向量 |
| 语义相似度通过向量距离衡量 | Semantic similarity is measured by vector distance | 语义相似度通过向量距离衡量 |
| 余弦相似度是最常用的度量 | Cosine similarity is the most common measure | 余弦相似度是最常用的度量 |
| 查询和文档必须用同一个 Embedding 模型 | Query and documents must use the SAME embedding model | 查询和文档必须用同一个 Embedding 模型 |
| OpenAI 限制单请求总 token 300,000 | OpenAI limits total tokens to 300,000 per request | OpenAI 限制单请求总 token 300,000 |
| 上下文 Embedding 比静态 Embedding 更精准 | Contextual embeddings are more accurate than static | 上下文 Embedding 比静态 Embedding 更精准 |
| Embedding 是 RAG 检索质量的决定因素 | Embedding quality determines RAG retrieval quality | Embedding 质量决定 RAG 检索质量 |
LCEL & Runnable
What is LCEL & Runnable?
- Runnable is a standard interface (protocol) that represents anything that can be “run” — it takes an input and produces an output. It is the basic building block of LangChain.
是一个标准接口（协议），代表任何“可运行”的东西——它接收输入并产生输出。它是 LangChain 的基础构建块 - LCEL (LangChain Expression Language) is a declarative way to compose chains using the pipe operator
|. It allows you to string together multiple Runnables into a processing pipeline
是一种声明式语言，通过管道操作符|来组合链。它允许你将多个 Runnable 串联成一个处理流水线
Core Content
Runnable Interface
Runnable is an abstract interface that defines five core methods.
| Method | Type | Purpose |
|---|---|---|
invoke(input) | 同步 | 单次调用，阻塞式返回完整结果 |
ainvoke(input) | 异步 | 单次异步调用，非阻塞 |
stream(input) | 同步 | 流式输出，逐 token 生成结果 |
batch([inputs]) | 同步 | 批量处理多个输入 |
abatch([inputs]) | 异步 | 异步批量处理 |
LCEL (LangChain Expression Language)
LCEL is the syntax/rules for composing Runnables. The core syntax is:
chain = component1 | component2 | component3
result = chain.invoke(input)Data flows left to right: output of component1 → input of component2 → output of component2 → input of component3
Key Runnable Primitives
Code Implementation
Basic LCEL Chain
# ============================================================
# 导入必要的模块
# Import necessary modules
# ============================================================
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# ============================================================
# 1. 创建各个组件（每个都实现了 Runnable 接口）
# 1. Create individual components (all implement Runnable interface)
# ============================================================
# 提示模板：定义与 LLM 对话的格式
# Prompt template: defines the format for talking to the LLM
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that translates {source_lang} to {target_lang}."),
("human", "{text}")
])
# LLM 模型：实际的语言模型
# LLM model: the actual language model
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 输出解析器：将 LLM 的字符串输出原样返回（或可解析为其他格式）
# Output parser: returns the LLM's string output as-is (or can parse to other formats)
parser = StrOutputParser()
# ============================================================
# 2. 用 LCEL 的管道操作符 | 组合成链
# 2. Compose into a chain using LCEL pipe operator |
# ============================================================
# 数据流：prompt -> model -> parser
# 每个组件的输出自动成为下一个组件的输入
# Data flow: prompt -> model -> parser
# Each component's output automatically becomes the next component's input
translation_chain = prompt | model | parser
# ============================================================
# 3. 调用链（invoke 是 Runnable 接口的标准方法）
# 3. Invoke the chain (invoke is the standard method of Runnable interface)
# ============================================================
result = translation_chain.invoke({
"source_lang": "English",
"target_lang": "Chinese",
"text": "Hello, how are you today?"
})
print(result)
# 输出: 你好，今天你好吗？Parallel Execution
# ============================================================
# RunnableParallel: 并行执行多个任务
# RunnableParallel: Execute multiple tasks in parallel
# ============================================================
from langchain_core.runnables import RunnableParallel
# ============================================================
# 定义两个独立的提示模板
# Define two independent prompt templates
# ============================================================
prompt_summary = ChatPromptTemplate.from_messages([
("system", "You are a summarizer. Summarize the following text in one sentence."),
("human", "{text}")
])
prompt_keywords = ChatPromptTemplate.from_messages([
("system", "You are a keyword extractor. Extract 3 keywords from the following text."),
("human", "{text}")
])
# ============================================================
# 创建两个独立的链
# Create two independent chains
# ============================================================
summary_chain = prompt_summary | model | parser
keywords_chain = prompt_keywords | model | parser
# ============================================================
# RunnableParallel 并行执行两个链
# RunnableParallel executes both chains in parallel
# ============================================================
# 注意：两个链共享同一个输入 {"text": ...}
# Note: Both chains share the same input {"text": ...}
parallel_chain = RunnableParallel({
"summary": summary_chain,
"keywords": keywords_chain
})
# ============================================================
# 调用并行链 - 两个任务同时执行
# Invoke the parallel chain - both tasks execute simultaneously
# ============================================================
result = parallel_chain.invoke({
"text": "LangChain is a framework for developing applications powered by language models. "
"It provides tools for building chains, agents, and retrieval systems."
})
print(f"Summary: {result['summary']}")
print(f"Keywords: {result['keywords']}")RunnablePassthrough
# ============================================================
# RunnablePassthrough: 传递数据或进行转换
# RunnablePassthrough: Pass data through or transform it
# ============================================================
from langchain_core.runnables import RunnablePassthrough
# ============================================================
# 场景：在输入进入 LLM 之前，对输入进行预处理
# Scenario: Preprocess the input before it goes into the LLM
# ============================================================
# 方式1：RunnablePassthrough 原样传递数据
# Method 1: RunnablePassthrough passes data through unchanged
# 这里用 RunnablePassthrough() 占位，表示"把输入原样传给下一步"
# Here RunnablePassthrough() is a placeholder meaning "pass input to next step as-is"
passthrough_chain = (
RunnablePassthrough() # 输入原样传递 / Pass input through
| prompt # 然后进入 prompt / Then into prompt
| model # 然后进入 model / Then into model
| parser # 最后解析 / Finally parse
)
# ============================================================
# 方式2：用字典 + lambda 函数进行数据转换
# Method 2: Use dict + lambda functions for data transformation
# ============================================================
# RunnablePassthrough.assign() 可以在传递数据的同时添加/修改字段
# RunnablePassthrough.assign() can add/modify fields while passing data
from langchain_core.runnables import RunnablePassthrough
# 假设我们想：如果输入文本超过100字符，就截断
# Suppose we want to: truncate input text if it exceeds 100 characters
preprocessing_chain = (
{
# 对 "text" 字段应用转换：如果超过100字符则截断
# Apply transformation to "text" field: truncate if > 100 chars
"text": lambda x: x["text"][:100] + "..." if len(x["text"]) > 100 else x["text"],
# "source_lang" 和 "target_lang" 原样传递
# "source_lang" and "target_lang" pass through unchanged
"source_lang": lambda x: x["source_lang"],
"target_lang": lambda x: x["target_lang"],
}
| prompt
| model
| parser
)RunnableBranch
# ============================================================
# RunnableBranch: 基于条件选择不同的执行路径
# RunnableBranch: Choose different execution paths based on conditions
# ============================================================
from langchain_core.runnables import RunnableBranch
from langchain_core.prompts import ChatPromptTemplate
# ============================================================
# 定义不同场景的提示模板
# Define prompt templates for different scenarios
# ============================================================
prompt_short = ChatPromptTemplate.from_messages([
("system", "You are a concise assistant. Answer briefly in under 20 words."),
("human", "{text}")
])
prompt_long = ChatPromptTemplate.from_messages([
("system", "You are a detailed assistant. Provide comprehensive answers."),
("human", "{text}")
])
# ============================================================
# 创建对应的链
# Create corresponding chains
# ============================================================
short_chain = prompt_short | model | parser
long_chain = prompt_long | model | parser
# ============================================================
# RunnableBranch: (条件, 链) 对 的列表
# RunnableBranch: list of (condition, chain) pairs
# ============================================================
# 如果文本长度 < 50 字符，用 short_chain，否则用 long_chain
# If text length < 50 chars, use short_chain, otherwise use long_chain
branch_chain = RunnableBranch(
(lambda x: len(x["text"]) < 50, short_chain), # 条件1：短文本 / Condition 1: short text
long_chain # 默认分支 / Default branch
)
# ============================================================
# 测试：短文本走 short_chain，长文本走 long_chain
# Test: short text goes to short_chain, long text goes to long_chain
# ============================================================
short_result = branch_chain.invoke({"text": "What is AI?"})
long_result = branch_chain.invoke({"text": "Explain the complete history of artificial intelligence from 1950 to today."})Key Takeaways
async await LLM call
LLM APIs are I/O-bound operations — your program spends most of its time waiting for the network response from the API server, not doing computation. In synchronous code, each LLM call blocks the entire thread until the response returns. With async/await, the event loop can handle other tasks while waiting for the response, dramatically improving throughput.
LLM API是I/O密集型操作——你的程序大部分时间都在等待API服务器的网络响应，而不是在进行计算。在同步代码中，每次LLM调用都会阻塞整个线程直到响应返回。使用async/await后，事件循环可以在等待响应时处理其他任务，大幅提升吞吐量。
Core Concepts
async / await
async def declares a coroutine function. When called, it returns a coroutine object, not the result immediately. await suspends the coroutine until the awaited operation completes, allowing the event loop to run other tasks.async def声明一个协程函数。调用时返回一个协程对象，而不是立即返回结果。await挂起协程直到被等待的操作完成，让事件循环可以运行其他任务。
import asyncio
# async def 声明这是一个协程函数
# async def declares this as a coroutine function
async def fetch_llm_response(prompt: str):
"""
模拟LLM API调用 - 等待2秒后返回结果
Simulate an LLM API call - wait 2 seconds then return result
"""
# await 挂起当前协程，让事件循环处理其他任务
# await suspends the current coroutine,
# allowing event loop to handle other tasks
await asyncio.sleep\(2\) # 模拟网络延迟 / Simulate network latency
# sleep\(2\) <-- '\' is to let wordpass correctly save it, nothing else.
# 这个 '\(' 奇怪的写法是为了网站 wordpress 能正确存储页面，和python无关
return f"Response to: {prompt}"
async def main():
# 调用async函数返回协程对象，需要await才能获取结果
# Calling an async function returns a coroutine object; need await to get result
result = await fetch_llm_response("Hello")
print(result)
# asyncio.run() 创建事件循环并运行main协程
# asyncio.run() creates an event loop and runs the main coroutine
asyncio.run(main())
Event Loop
The event loop is the core of asyncio. It manages and schedules all coroutines, switching between them when they hit await points. Think of it as a traffic controller — when one task is waiting for I/O, it switches to another task that’s ready to run.
事件循环是asyncio的核心。它管理和调度所有协程，在它们遇到await点时进行切换。可以把它想象成一个交通调度员——当一个任务在等待I/O时，它切换到另一个可以运行的任务
import asyncio
import time
async def task(name: str, delay: float):
print(f"[{time.strftime('%H:%M:%S')}] Task {name}: starting")
await asyncio.sleep(delay) # 模拟I/O等待 / Simulate I/O wait
print(f"[{time.strftime('%H:%M:%S')}] Task {name}: done")
return f"Result from {name}"
async def main():
# 并发执行三个任务 - 总耗时约2秒而非6秒
# Run three tasks concurrently - total ~2s not 6s
results = await asyncio.gather(
task("A", 2),
task("B", 1),
task("C", 1.5)
)
print(f"All results: {results}")
asyncio.run(main())
# 输出示例 / Example output:
# [14:30:01] Task A: starting
# [14:30:01] Task B: starting
# [14:30:01] Task C: starting
# [14:30:02] Task B: done
# [14:30:02.5] Task C: done
# [14:30:03] Task A: done
# All results: ['Result from A', 'Result from B', 'Result from C']OpenAI Async Calls
Basic Async Call
基础异步调用
import asyncio
import os
from openai import AsyncOpenAI # 异步客户端 / Async client
# ============================================================
# 1. 初始化异步客户端
# 1. Initialize async client
# ============================================================
# AsyncOpenAI 提供与OpenAI相同的接口，但所有方法都是异步的
# AsyncOpenAI provides the same interface as OpenAI, but all methods are async
# 推荐从环境变量读取API Key / Recommended to read API Key from env var
client = AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
# timeout=60.0, # 可配置超时 / Optional timeout
# max_retries=2, # 可配置重试 / Optional retries
)
# ============================================================
# 2. 定义异步调用函数
# 2. Define async call function
# ============================================================
async def get_chat_completion(prompt: str, model: str = "gpt-3.5-turbo") -> str:
"""
异步获取LLM响应
Asynchronously get LLM response
Args:
prompt: 用户提示词 / User prompt
model: 模型名称 / Model name
Returns:
LLM响应内容 / LLM response content
"""
try:
# await client.chat.completions.create 是非阻塞的
# await client.chat.completions.create is non-blocking
# 发起HTTP请求后，事件循环可以处理其他任务
# After initiating HTTP request, event loop can handle other tasks
response = await client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=500,
)
# 提取响应内容 / Extract response content
# response.choices[0].message.content 包含完整响应
# response.choices[0].message.content contains the full response
return response.choices[0].message.content
except Exception as e:
# 捕获并处理异常 / Catch and handle exceptions
print(f"Error calling OpenAI API: {e}")
return f"Error: {str(e)}"
# ============================================================
# 3. 执行异步调用
# 3. Execute async call
# ============================================================
async def main():
# 单个调用 / Single call
result = await get_chat_completion("What is the capital of France?")
print(f"Response: {result}")
# 启动事件循环 / Start the event loop
if __name__ == "__main__":
asyncio.run(main())Concurrent Multiple Calls
并发调用多个LLM
import asyncio
import os
import time
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
async def summarize_text(text: str, idx: int) -> dict:
"""
异步总结单条文本
Asynchronously summarize a single text
"""
start = time.time()
try:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "Summarize the following text in one sentence."},
{"role": "user", "content": text}
],
max_tokens=100,
)
elapsed = time.time() - start
return {
"index": idx,
"summary": response.choices[0].message.content,
"elapsed": elapsed
}
except Exception as e:
return {
"index": idx,
"summary": f"Error: {e}",
"elapsed": time.time() - start
}
async def main():
# 模拟10条需要总结的文本 / Simulate 10 texts to summarize
texts = [
f"This is sample text number {i}. " * 5
for i in range(10)
]
print(f"Starting {len(texts)} concurrent summarization requests...")
start_total = time.time()
# ============================================================
# asyncio.gather() 并发执行所有协程
# asyncio.gather() runs all coroutines concurrently
# ============================================================
# 所有请求同时发出，总耗时 ≈ 最慢单次请求的耗时
# All requests are sent simultaneously; total time ≈ slowest single request
results = await asyncio.gather(*[
summarize_text(text, i) for i, text in enumerate(texts)
])
total_elapsed = time.time() - start_total
print(f"\nAll {len(results)} requests completed in {total_elapsed:.2f}s")
# 打印结果 / Print results
for r in results:
print(f" [{r['index']}] {r['summary'][:50]}... ({r['elapsed']:.2f}s)")
if __name__ == "__main__":
asyncio.run(main())Streaming
流式输出
import asyncio
import os
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
async def stream_response(prompt: str):
"""
流式获取LLM响应 - 逐词/逐块输出
Stream LLM response - output token by token / chunk by chunk
"""
# ============================================================
# stream=True 启用流式模式
# stream=True enables streaming mode
# ============================================================
# 注意：stream=True时，需要使用 async for 迭代响应流
# Note: when stream=True, use async for to iterate the response stream
stream = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "user", "content": prompt}
],
stream=True, # 启用流式 / Enable streaming
max_tokens=300,
)
print("Streaming response:")
full_content = ""
# ============================================================
# async for 逐个处理流式chunk
# async for processes each streaming chunk one by one
# ============================================================
# 每个chunk包含一小部分响应内容
# Each chunk contains a small piece of the response
async for chunk in stream:
# delta 是当前chunk的新增内容
# delta is the new content in this chunk
delta = chunk.choices[0].delta
if delta.content:
content = delta.content
print(content, end="", flush=True) # 实时输出 / Real-time output
full_content += content
print("\n\n--- Full response received ---")
return full_content
async def main():
await stream_response("Explain quantum computing in simple terms.")
if __name__ == "__main__":
asyncio.run(main())Error Handling & Retry
import asyncio
import os
import random
from openai import AsyncOpenAI, APIError, APIConnectionError, RateLimitError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# ============================================================
# 使用 tenacity 库实现指数退避重试
# Use tenacity library for exponential backoff retry
# ============================================================
# 重试策略：
# - 最多重试3次 / Max 3 retry attempts
# - 等待时间：2秒 × 2^(attempt-1) + 随机抖动
# - Wait time: 2s × 2^(attempt-1) + random jitter
# - 仅对特定异常类型重试 / Only retry on specific exception types
@retry(
stop=stop_after_attempt(3), # 最多3次尝试 / Max 3 attempts
wait=wait_exponential(multiplier=2, min=2, max=30), # 指数退避 / Exponential backoff
retry=retry_if_exception_type((APIError, APIConnectionError, RateLimitError)),
)
async def robust_chat_completion(prompt: str) -> str:
"""
带自动重试的LLM调用
LLM call with automatic retry
"""
# 模拟偶尔的网络错误 / Simulate occasional network errors
if random.random() < 0.3: # 30%概率触发重试 / 30% chance to trigger retry
raise APIConnectionError("Simulated network error")
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
max_tokens=100,
)
return response.choices[0].message.content
async def main():
try:
# 即使有30%概率失败，重试机制会保证最终成功
# Even with 30% failure rate, retry mechanism ensures eventual success
result = await robust_chat_completion("Hello, world!")
print(f"Result: {result}")
except Exception as e:
print(f"All retries exhausted: {e}")
if __name__ == "__main__":
asyncio.run(main())Production Best Practices
生产环境最佳实践
import asyncio
import os
from contextlib import asynccontextmanager
from openai import AsyncOpenAI
# ============================================================
# 1. 客户端单例模式 (Singleton Client)
# ============================================================
# 整个应用共享一个AsyncOpenAI实例，复用连接池
# Share one AsyncOpenAI instance across the app to reuse connection pool
class LLMClient:
_instance = None
@classmethod
def get_client(cls) -> AsyncOpenAI:
if cls._instance is None:
cls._instance = AsyncOpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
timeout=60.0,
max_retries=2,
# httpx客户端配置 / httpx client config
http_client=None, # 可自定义 / Can customize
)
return cls._instance
# ============================================================
# 2. 上下文管理器自动清理 (Context Manager for Cleanup)
# ============================================================
@asynccontextmanager
async def get_llm_client():
"""
异步上下文管理器，确保客户端正确关闭
Async context manager to ensure proper client cleanup
"""
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
try:
yield client
finally:
# 关闭底层httpx连接 / Close underlying httpx connections
await client.close()
async def use_client():
async with get_llm_client() as client:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello"}],
)
return response.choices[0].message.content
# ============================================================
# 3. 超时控制 (Timeout Control)
# ============================================================
async def call_with_timeout(prompt: str, timeout_seconds: float = 30.0):
"""
带超时的LLM调用
LLM call with timeout
"""
try:
# asyncio.timeout 会在超时时抛出 TimeoutError
# asyncio.timeout raises TimeoutError on timeout
async with asyncio.timeout(timeout_seconds):
client = LLMClient.get_client()
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content
except asyncio.TimeoutError:
return f"Request timed out after {timeout_seconds}s"
except Exception as e:
return f"Error: {e}"
async def main():
# 测试超时 / Test timeout
result = await call_with_timeout("Tell me a very long story...", timeout_seconds=5.0)
print(result)
if __name__ == "__main__":
asyncio.run(main())Key Takeaways
| 要点 | EN | CN |
|---|---|---|
使用 AsyncOpenAI 替代 OpenAI | Use AsyncOpenAI instead of OpenAI | 用 AsyncOpenAI 替换 OpenAI |
所有API调用前加 await | Add await before every API call | 每个API调用前加 await |
asyncio.gather() 实现并发 | Use asyncio.gather() for concurrency | 用 asyncio.gather() 实现并发 |
流式输出用 async for | Use async for for streaming | 流式输出用 async for |
| 生产环境需处理超时和重试 | Handle timeouts and retries in production | 生产环境需处理超时和重试 |
| 整个应用共享一个客户端实例 | Share one client instance across the app | 整个应用共享一个客户端实例 |