LangChain框架在高炉炼铁智能化领域的应用~系列文章07：RAG检索增强生成 — 高炉知识库的“百科全书“

📚 第7期：RAG检索增强生成 — 高炉知识库的"百科全书"

专栏：《LangChain框架在高炉炼铁智能化领域的应用》
前情回顾：上期我们用 Chain 串起了 AI 处理流水线
本期重点：RAG（检索增强生成）—— 让 AI 学会"查资料"再回答问题

📖 引言：AI 的"知识瓶颈"

想象一个场景 👇

你问 AI：

👨‍🏭 你：“5号高炉在2023年3月那次悬料事故是怎么处理的？”

🤖 AI（自信满满）：“那场事故是因为风量突然下降导致的，建议减风加焦……”

然后你翻出了档案记录：

📄 档案记录：“2023年3月悬料事故根本原因是原燃料质量波动，执行了排风+空吹操作……”

😱 AI 完全说错了！

大模型的问题：

• ❌ 训练数据截止日期之后的信息不知道
• ❌ 企业内部专有数据没见过
• ❌ 会"幻觉"——编造看似合理但错误的内容

解决方案 = RAG（Retrieval-Augmented Generation） 🦸‍♂️

🧩 RAG 是什么？三句话

RAG = 检索（Retrieve） + 增强（Augment） + 生成（Generate）

1️⃣ 检索：从知识库中找到相关的文档片段
2️⃣ 增强：把检索到的内容拼接到 Prompt 中
3️⃣ 生成：让 AI 基于这些资料回答问题

核心原理图：

用户问题 ──→ [检索] ──→ 向量知识库
│ ↑
│ ┌────┴────┐
│ │ 文档分片 │
│ │ 向量化 │
▼ └─────────┘
[增强 Prompt]
│
[生成回答]
▼
最终答案 ✅

类比高炉场景：

没有RAG的AI → 凭经验开高炉的老师傅（记忆有限、会忘事）
有RAG的AI → 带着操作手册开高炉的老师傅（随时翻书、准确率高）

🏗️ 搭建RAG系统的完整流程（9步详解）

Step 1~4：文档处理（构建知识库）

# 📁 rag_knowledge_base.py
# 高炉知识库构建脚本

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from typing import List, Dict
import os

# ─────────── Step 1: 加载文档 ───────────

def load_blast_furnace_documents() –> List[str]:
"""
加载高炉相关的知识文档
实际场景中，这些文档来自：操作手册、技术标准、历史案例、论文等
"""
documents = [
# 模拟知识文档片段
"""
【高炉悬料处理规程】
当出现悬料征兆时（风压升高>0.45MPa，风量下降>10%）：
第一步：立即减风 10-15%，稳定压量关系
第二步：如 5 分钟后无改善，进一步减风至 50%
第三步：如仍未崩料，执行排风操作
注意事项：严禁在悬料状态下强行加风！
""",
"""
【炉温判断标准】
铁水温度正常范围：1480-1520°C
硅含量正常范围：0.35-0.55%
炉渣碱度正常范围：1.10-1.25
当铁温<1480°C或硅含量<0.35%时，判定为"炉温偏低"
当铁温>1520°C或硅含量>0.55%时，判定为"炉温偏高"
""",
"""
【布料矩阵调整原则】
正常生产时使用"平台+漏斗"布料模式
边缘发展时，适当增加边缘矿量
中心发展时，适当增加中心焦量
布料角度通常控制在 8°-45° 之间
"""
]
return documents

# ─────────── Step 2: 文档分割 ───────────

def split_documents(documents: List[str]) –> List:
"""将长文档切分成合适的片段"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每段500字符
chunk_overlap=50, # 重叠50字符，避免切碎关键信息
separators=["\\n\\n", "\\n", "。", ".", " ", ""],
length_function=len
)

chunks = []
for doc in documents:
splits = text_splitter.split_text(doc)
chunks.extend(splits)

print(f"📄 原始文档: {len(documents)} 篇")
print(f"✂️ 分割后: {len(chunks)} 个片段")
return chunks

# ─────────── Step 3: 向量化存储 ───────────

def create_vector_store(chunks: List[str], persist_dir: str = "./bf_knowledge_db"):
"""将文档片段向量化并存入向量数据库"""
embeddings = OpenAIEmbeddings(
model="doubao-seed-2-0-lite-260215", # 支持向量化的模型
api_key=os.getenv("COZE_WORKLOAD_IDENTITY_API_KEY"),
base_url=os.getenv("COZE_INTEGRATION_MODEL_BASE_URL"),
)

# 创建向量存储
vector_store = Chroma.from_texts(
texts=chunks,
embedding=embeddings,
persist_directory=persist_dir
)
# 持久化保存
vector_store.persist()

print(f"💾 向量库已保存至: {persist_dir}")
print(f"📊 向量维度: {len(vector_store.get()[0]) if hasattr(vector_store,'get') else 'N/A'}")
return vector_store

# ─────────── 执行构建 ───────────

if __name__ == "__main__":
print("🚀 开始构建高炉知识库…")
docs = load_blast_furnace_documents()
chunks = split_documents(docs)
vector_store = create_vector_store(chunks)
print("✅ 知识库构建完成！")

Step 5~9：RAG 问答系统

# 📁 rag_qa_system.py
# 基于RAG的高炉知识问答系统

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import os

# ─────────── Step 5: 加载知识库 ───────────

def load_vector_store(persist_dir: str = "./bf_knowledge_db"):
"""加载已持久化的向量库"""
embeddings = OpenAIEmbeddings(
model="doubao-seed-2-0-lite-260215",
api_key=os.getenv("COZE_WORKLOAD_IDENTITY_API_KEY"),
base_url=os.getenv("COZE_INTEGRATION_MODEL_BASE_URL"),
)
return Chroma(
persist_directory=persist_dir,
embedding_function=embeddings
)

# ─────────── Step 6: 构建检索器 ───────────

def build_retriever(vector_store, k: int = 3):
"""
构建检索器
k=3：每次检索返回最相关的3个文档片段
"""
return vector_store.as_retriever(
search_type="similarity", # 相似度检索
search_kwargs={"k": k} # 返回Top-K结果
)

# ─────────── Step 7: 设计 RAG Prompt ───────────

rag_prompt_template = """你是一名高炉炼铁专家。请基于以下参考资料回答用户的问题。

【参考资料】
{context}

【用户问题】
{question}

回答要求：
1. 优先使用参考资料中的内容回答
2. 如果参考资料不足以回答，请明确告知
3. 引用参考资料时说明来源
4. 回答要专业、具体、可操作

【回答】"""

rag_prompt = PromptTemplate(
template=rag_prompt_template,
input_variables=["context", "question"]
)

# ─────────── Step 8: 构建 RAG Chain ───────────

def build_rag_chain(vector_store):
"""构建完整的RAG问答链"""
llm = ChatOpenAI(
model="doubao-seed-2-0-lite-260215",
temperature=0.2, # 低温度，严格基于资料回答
timeout=600
)

retriever = build_retriever(vector_store, k=3)

# 使用 LangChain 内置的 RetrievalQA
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # 把所有检索结果一起塞给LLM
retriever=retriever,
chain_type_kwargs={
"prompt": rag_prompt,
"verbose": True # 显示检索过程
},
return_source_documents=True # 返回检索到的原文
)
return qa_chain

# ─────────── Step 9: 执行问答 ───────────

def ask_rag(qa_chain, question: str):
"""向RAG系统提问"""
result = qa_chain.invoke({"query": question})

print(f"\\n{'='*50}")
print(f"❓ 问题: {question}")
print(f"{'='*50}")
print(f"🤖 回答:\\n{result['result']}")

print(f"\\n📚 参考来源 ({len(result['source_documents'])} 篇):")
for i, doc in enumerate(result['source_documents'], 1):
print(f" [{i}] {doc.page_content[:100]}…")

return result

# ─────────── 运行 ───────────

if __name__ == "__main__":
print("🚀 初始化RAG系统…")
vector_store = load_vector_store()
qa_chain = build_rag_chain(vector_store)

# 测试几个问题
questions = [
"铁水温度低于1480°C应该怎么办？",
"高炉悬料怎么处理？",
"布料矩阵调整的原则是什么？"
]

for q in questions:
ask_rag(qa_chain, q)
print("\\n")

🔍 RAG vs 纯LLM：对比实验

我们用一个真实的高炉场景问题来对比：

问题

“铁水温度 1475°C，硅含量 0.32%，应该怎么调整？”

结论：工业场景强烈推荐 RAG！ ✅✅✅

🏭 实战进阶：高炉操作案例库RAG

构建一个高炉历史操作案例的 RAG 系统，让 AI 能从历史事故中学习 🔥

# 模拟一个高炉历史案例数据结构
historical_cases = [
"""
案例编号：CASE-2023-0315
时间：2023年3月15日
高炉：4号高炉（3200m³）
事件类型：悬料
前兆：风压在30分钟内从0.38MPa升至0.46MPa
风量从5200m³/min降至4700m³/min
透气性指数从35降至26
处理措施：1. 立即减风15%
2. 减风后5分钟无改善，二次减风至50%
3. 执行排风操作，炉料崩落
4. 恢复风量至80%，观察15分钟后逐步加风
结果：成功处理，未造成设备损坏
经验总结：悬料初期果断减风是成功的关键
""",
"""
案例编号：CASE-2024-0108
时间：2024年1月8日
高炉：5号高炉（2500m³）
事件类型：炉凉
前兆：铁水温度从1510°C连续8小时降至1465°C
硅含量从0.48%降至0.30%
炉渣颜色变黑
处理措施：1. 加焦总量15吨，分批加入
2. 适当降低风量5%
3. 检查冷却设备，发现漏水
4. 处理漏水后，炉温逐步恢复
结果：48小时后炉温恢复正常
经验总结：炉凉要查冷却系统是否有漏水
"""
]

# 将这些案例也向量化存入知识库
# 这样AI就能在遇到类似问题时，"回忆"起历史案例的处置方法！

⚡ RAG优化技巧（高炉场景专用）

1️⃣ 检索策略优化

# 混合检索：相似度 + 关键词
retriever = vector_store.as_retriever(
search_type="mmr", # Maximum Marginal Relevance
search_kwargs={
"k": 5,
"fetch_k": 20, # 先取20个候选
"lambda_mult": 0.5 # 多样性系数（0~1）
}
)

2️⃣ 文档重排序

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 用LLM对检索结果进行"精炼"——只保留最相关内容
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)

3️⃣ 针对高炉场景的文档分片策略

# 高炉文档的特殊分片：按"主题"切分，而非简单按字符
bf_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=[
"\\n## ", # 一级标题
"\\n### ", # 二级标题
"\\n【", # 高炉规范常用【】标记
"\\n案例编号：", # 案例库标记
"\\n\\n", "\\n", " ", ""
]
)

📊 本期小结

核心心法：

RAG 是工业 AI 的"基础设施"——没有 RAG 的 AI 是"纸上谈兵"，有 RAG 的 AI 是"实战专家"。在要求高准确率的高炉场景中，RAG 不是可选项，而是必选项！

📌 下期预告

第8期：《Agents智能体：给高炉装上"自主决策大脑"》 🤖

Chain 学会了按步骤执行，RAG 学会了查资料——但还不够！

真正的工业场景中，AI 需要自主判断：什么时候该查数据？什么时候该调知识库？什么时候该调用计算工具？甚至——该不该拉响警报？

这就是 Agent（智能体） 的威力！它不再是"被动回答"，而是"主动思考、自主行动"！

下一期，我们进入整个专栏的高潮部分——打造一个能自主决策、能操作工具、能多步推理的高炉智能体！🚀

🌟 16期连载中，精彩内容不要错过！
作者：高炉炼铁智能化技术研究者，专注钢铁冶金与人工智能 交叉领域。

👍 如果觉得有帮助，请点赞、收藏、转发！
版权归作者所有，未经许可请勿抄袭，套用，商用(或其它具有利益性行为)。
🔔 关注专栏，不错过后续精彩内容！