1. 项目背景与核心价值
最近两年,大模型技术在企业应用领域快速落地,但很多团队在具体实施时都会遇到一个共性难题:如何将大语言模型(LLM)真正融入现有业务流程?我在三个不同规模的企业级项目中,都采用了LangChain和LlamaIndex这两个框架来构建智能工作流,今天就把实战中积累的经验做个系统梳理。
传统企业系统对接大模型时,往往会面临三大痛点:
- 业务数据分散在多个孤岛系统(CRM/ERP/OA等)
- 现有工作流需要保持稳定性
- 不同部门对AI能力的需求差异大
通过LangChain的模块化组件和LlamaIndex的高效检索能力,我们成功将GPT-4、Claude等大模型无缝嵌入到客户服务、智能审批、知识管理等20+业务场景。最典型的案例是某金融机构的智能投研系统,通过工作流编排将响应速度提升6倍,同时将人工复核工作量减少80%。
2. 技术架构设计
2.1 核心组件选型
在技术栈选择上,我们采用分层架构设计:
code复制[业务系统层]
│
▼
[工作流编排层] ← LangChain
│
▼
[数据接入层] ← LlamaIndex
│
▼
[大模型服务层] ← GPT-4/Claude/Mistral等
关键决策:没有选择直接调用大模型API,而是通过中间层实现业务逻辑与大模型解耦。这样当需要切换底层模型时,业务代码几乎不需要修改。
2.2 LangChain的核心作用
具体到代码层面,LangChain主要解决三类问题:
- 流程控制:通过SequentialChain实现多步骤决策
python复制from langchain.chains import SequentialChain
analysis_chain = SequentialChain(
chains=[data_clean_chain, classify_chain, report_chain],
input_variables=["raw_input"],
output_variables=["final_report"]
)
- 业务工具集成:用Tool接口封装企业现有系统
python复制from langchain.tools import Tool
crm_tool = Tool(
name="CRM查询",
func=query_crm_system,
description="根据客户ID查询历史订单"
)
- 条件路由:通过LLMRouterChain实现智能分流
python复制router_template = """根据用户问题类型选择处理方式:
{input}"""
router_chain = LLMRouterChain.from_llm(llm, router_template)
2.3 LlamaIndex的数据处理
企业数据通常分布在三个位置:
- 结构化数据库(MySQL/Oracle)
- 非结构化文档(PDF/Word)
- 第三方API(企查查/天眼查)
我们使用LlamaIndex构建统一的数据接入层:
python复制from llama_index import (
VectorStoreIndex,
SQLDatabase,
Document
)
# 处理PDF合同
documents = SimpleDirectoryReader("contracts/").load_data()
vector_index = VectorStoreIndex.from_documents(documents)
# 连接业务数据库
sql_index = SQLDatabase.from_uri("mysql://user:pass@host/db")
# 组合检索器
retriever = HybridRetriever(vector_index, sql_index)
3. 典型工作流实现
3.1 智能客服工单处理
以金融行业的投诉处理为例,完整流程包括:
- 工单分类(紧急/普通)
- 客户画像生成
- 历史相似案例检索
- 回复话术生成
- 合规性检查
对应的LangChain实现:
python复制complaint_chain = {
"input": lambda x: x["user_input"],
"category": classify_chain,
"profile": customer_chain,
"similar_cases": retriever_chain,
"draft": reply_chain,
"check": compliance_chain
} | RunnableParallel
# 加入人工复核节点
final_chain = complaint_chain | RunnableLambda(
lambda x: human_review(x) if x["category"] == "紧急" else x
)
3.2 合同智能审查
法律文档处理需要特别注意:
- 条款变更追踪
- 风险条款识别
- 版本差异比对
我们开发了专用的文档处理流程:
python复制class ContractAnalyzer:
def __init__(self):
self.parser = ClauseParser()
self.risk_model = load_risk_model()
def analyze(self, doc_path):
text = extract_text(doc_path)
clauses = self.parser.parse(text)
risks = []
for clause in clauses:
risk_score = self.risk_model.predict(clause.text)
if risk_score > 0.7:
risks.append({
"clause": clause.title,
"risk": risk_score,
"suggestion": generate_suggestion(clause)
})
return risks
4. 性能优化实践
4.1 检索增强生成(RAG)优化
在知识库场景下,我们通过以下手段提升效果:
- 分层索引:
- 第一层:元数据过滤(部门/日期/类型)
- 第二层:语义检索(向量相似度)
- 第三层:关键词召回(BM25)
python复制index = HierarchicalIndex(
metadata_filter=MetadataFilter(fields=["department","date"]),
vector_index=VectorStoreIndex(...),
keyword_index=BM25Index(...)
)
- 查询重写:
python复制def query_rewrite(original_query):
rewrite_prompt = f"""将用户查询改写成更适合检索的形式:
原始查询: {original_query}
改写后:"""
return llm(rewrite_prompt)
4.2 缓存策略
针对高频查询设计三级缓存:
- 内存缓存(最近5分钟结果)
- Redis缓存(当天热点问题)
- 磁盘缓存(历史标准答案)
python复制from langchain.cache import (
InMemoryCache,
RedisCache,
SQLiteCache
)
chain = load_chain().with_cache(
caches=[
InMemoryCache(ttl=300),
RedisCache(redis_url="redis://localhost"),
SQLiteCache(database_path=".cache.db")
]
)
5. 生产环境注意事项
5.1 安全合规要点
- 数据脱敏:
python复制from presidio_analyzer import AnalyzerEngine
analyzer = AnalyzerEngine()
def anonymize(text):
results = analyzer.analyze(text=text, language="zh")
for result in results:
text = text.replace(result.text, "[REDACTED]")
return text
- 审计日志:
python复制class AuditLogger:
def log(self, chain_input, chain_output):
record = {
"timestamp": datetime.now(),
"user": get_current_user(),
"input_hash": sha256(chain_input),
"model": chain_output["model"],
"cost": calculate_cost(chain_output)
}
insert_to_audit_db(record)
5.2 监控指标设计
必须监控的四大黄金指标:
- 延迟:P99 < 3s
- 准确率:关键步骤设置校验点
- 成本:按业务线拆分计算
- 稳定性:失败请求自动重试
Prometheus监控示例:
python复制from prometheus_client import Summary
REQUEST_TIME = Summary('request_processing_seconds', 'Time spent processing request')
@REQUEST_TIME.time()
def process_request(input):
# 业务逻辑
6. 踩坑经验分享
6.1 中文处理特别注意事项
- 分词问题:
- 法律/医疗等专业领域需要加载自定义词典
- 重要术语要设置同义词映射表
- 标点符号处理:
python复制def clean_chinese_text(text):
# 统一全角标点
text = text.replace(",", ",").replace("。", ".")
# 移除特殊空白符
return re.sub(r"[\u3000\xa0]", " ", text)
6.2 模型微调策略
对于专业领域任务,建议采用:
- Lora微调基础模型
- Prompt-tuning优化关键步骤
- 业务数据持续强化学习
python复制from peft import LoraConfig
lora_config = LoraConfig(
r=8,
target_modules=["q_proj", "v_proj"],
task_type="CAUSAL_LM"
)
在实际项目中,我们发现工作流中不同节点对模型要求差异很大。比如分类任务可以用较小的模型(如GPT-3.5),而生成任务则需要更强的模型(如GPT-4)。通过LangChain的RunnableBranch可以实现智能路由:
python复制from langchain.schema.runnable import RunnableBranch
model_router = RunnableBranch(
(lambda x: x["task_type"] == "分类", gpt3_chain),
(lambda x: x["task_type"] == "生成", gpt4_chain),
default_chain
)