LangChain链式调用：构建复杂的多步决策流程-编程阁

LangChain链式调用：构建复杂的多步决策流程

核心价值：通过LangChain的链式调用机制，将多个AI能力模块（如图像识别、自然语言理解）串联成具备逻辑推理能力的自动化流程，实现从“单点智能”到“系统智能”的跃迁。

在当前大模型应用开发中，单一模型往往难以应对复杂业务场景。例如，在一个智能客服系统中，用户可能上传一张产品故障图片并附带文字描述。系统不仅需要识别图像内容，还需理解用户情绪、提取关键问题，并生成结构化响应建议。这就要求我们构建多步决策流程，而LangChain提供的Chain机制正是解决此类问题的核心工具。

本文将以阿里开源的“万物识别-中文-通用领域”模型为基础，结合PyTorch环境下的图像推理能力，演示如何使用LangChain将图像识别与后续文本处理环节进行链式编排，打造一个可扩展的智能决策系统。

万物识别-中文-通用领域：技术背景与集成准备

“万物识别-中文-通用领域”是阿里巴巴推出的一款面向中文场景的通用图像识别模型，支持对日常物体、场景、文字等多类信息的细粒度识别，并输出带有语义标签的结果。该模型基于大规模中文图文对训练，在中文语境下的标签准确性和可读性显著优于传统英文模型。

技术优势与适用场景

✅中文原生支持：标签体系以中文组织，无需二次翻译
✅通用性强：覆盖超过1万种常见类别
✅轻量高效：可在本地GPU环境下快速部署（依赖PyTorch 2.5）
✅可解释性高：输出包含置信度和层级分类路径

该模型特别适用于： - 智能客服中的图片问题分析 - 内容审核中的图文一致性判断 - 零售行业的商品自动识别 - 教育领域的作业图像解析

环境配置与基础推理实现

首先确保运行环境已正确配置：

# 激活指定conda环境 conda activate py311wwts # 查看依赖（位于/root目录） pip install -r /root/requirements.txt

假设requirements.txt中已包含以下关键依赖：

torch==2.5.0 transformers Pillow langchain-core langchain-community

接下来，创建基础推理脚本推理.py，用于加载模型并执行单张图片识别。

# 推理.py from PIL import Image import torch from transformers import AutoModel, AutoTokenizer # 加载预训练模型和分词器 model_name = "bailing-model" # 假设本地已下载或可通过hub访问 model = AutoModel.from_pretrained(model_name) tokenizer = AutoTokenizer.from_pretrained(model_name) def predict(image_path: str) -> dict: image = Image.open(image_path).convert("RGB") # 图像预处理（根据模型要求调整） inputs = tokenizer(images=image, return_tensors="pt") with torch.no_grad(): outputs = model(**inputs) # 解码预测结果（简化示例） labels = outputs.logits.argmax(dim=-1).tolist() confidence = torch.softmax(outputs.logits, dim=-1).max(dim=-1).values.item() result = { "labels": [f"标签_{i}" for i in labels], # 实际应映射为真实中文标签 "confidence": confidence, "raw_output": outputs.logits.tolist() } return result if __name__ == "__main__": result = predict("/root/bailing.png") print("识别结果：", result)

⚠️ 注意：实际部署时需根据官方文档替换正确的模型加载方式和输入格式。若模型未公开发布，可能需要联系阿里获取授权或使用其API服务。

你可以将脚本复制到工作区以便编辑：

cp 推理.py /root/workspace cp bailing.png /root/workspace

并修改predict()函数中的路径为/root/workspace/bailing.png。

构建LangChain链式流程：从图像识别到语义决策

现在我们将上述图像识别功能封装为LangChain的一个组件，并与其他NLP模块组合成一个多步决策链。

第一步：定义自定义ImageRecognitionChain

LangChain允许我们通过继承BaseChain来创建自定义链。我们将图像识别过程包装为一个可复用的链节点。

# chains/image_recognition_chain.py from langchain_core.chains import Chain from typing import Dict, Any from PIL import Image import torch class ImageRecognitionChain(Chain): """封装万物识别模型为LangChain可调用链""" model: Any tokenizer: Any input_key: str = "image_path" output_key: str = "vision_result" @property def input_keys(self) -> list[str]: return [self.input_key] @property def output_keys(self) -> list[str]: return [self.output_key] def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]: image_path = inputs[self.input_key] image = Image.open(image_path).convert("RGB") inputs_tensor = self.tokenizer(images=image, return_tensors="pt") with torch.no_grad(): outputs = self.model(**inputs_tensor) # 简化后处理逻辑 predicted_label_id = outputs.logits.argmax(dim=-1).item() confidence = torch.softmax(outputs.logits, dim=-1).max().item() label_map = {0: "电子产品", 1: "家具", 2: "食品", 3: "衣物"} # 示例映射 detected_category = label_map.get(predicted_label_id, "未知") return { self.output_key: { "category": detected_category, "confidence": round(confidence, 4), "raw_logits": outputs.logits.tolist() } } @classmethod def from_pretrained(cls, model_name: str): model = AutoModel.from_pretrained(model_name) tokenizer = AutoTokenizer.from_pretrained(model_name) return cls(model=model, tokenizer=tokenizer)

第二步：构建下游文本决策链

假设我们希望根据图像识别结果生成客户回复建议。我们可以使用另一个LLM链来完成此任务。

# chains/response_generation_chain.py from langchain_core.prompts import PromptTemplate from langchain_community.llms import HuggingFacePipeline from langchain_core.chains import LLMChain # 模拟本地部署的中文LLM（如Qwen、ChatGLM） llm = HuggingFacePipeline.from_model_id( model_id="qwen/Qwen-7B-Chat", task="text-generation", device=0 # 使用GPU ) prompt = PromptTemplate( input_variables=["category", "confidence"], template=""" 你是一个智能客服助手。请根据图片识别结果生成一段友好且专业的回复建议。 识别类别：{category} 置信度：{confidence:.2f} 请用中文生成一条不超过80字的客户回复建议： """ ) response_chain = LLMChain(llm=llm, prompt=prompt)

第三步：串联形成完整多步决策链

使用LangChain的SequentialChain将两个独立链连接起来，形成端到端流程。

# main_pipeline.py from langchain_core.chains import SequentialChain from chains.image_recognition_chain import ImageRecognitionChain from chains.response_generation_chain import response_chain # 初始化图像识别链 vision_chain = ImageRecognitionChain.from_pretrained("bailing-model") # 构建总链 overall_chain = SequentialChain( chains=[vision_chain, response_chain], input_variables=["image_path"], output_variables=["text"], # 来自response_chain的输出 verbose=True ) # 执行全流程 result = overall_chain.invoke({ "image_path": "/root/workspace/bailing.png" }) print("最终输出：", result["text"])

运行结果示例：

识别到用户上传的图片属于“电子产品”，置信度0.96。 生成回复建议：您好，检测到您上传的是电子设备图片，请问是否遇到开机异常或屏幕损坏问题？我们将尽快为您处理。

多路径决策设计：基于条件分支的智能路由

在真实系统中，不同识别结果应触发不同的处理流程。LangChain虽不直接支持条件跳转，但我们可以通过自定义逻辑实现动态链路由。

# chains/routing_chain.py class ConditionalRoutingChain(Chain): vision_chain: ImageRecognitionChain electronic_chain: LLMChain food_chain: LLMChain default_chain: LLMChain def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]: image_path = inputs["image_path"] # 第一步：图像识别 vision_result = self.vision_chain.run({"image_path": image_path}) category = vision_result["category"] # 第二步：根据类别选择处理链 if category == "电子产品": advice = self.electronic_chain.run(vision_result) elif category == "食品": advice = self.food_chain.run(vision_result) else: advice = self.default_chain.run(vision_result) return { "category": category, "confidence": vision_result["confidence"], "response_advice": advice.strip() }

这种模式实现了真正的多步决策流，可根据中间结果动态选择后续动作，适用于： - 客服工单自动分类 - 医疗影像初筛后的专家转诊建议 - 自动化质检系统的缺陷等级判定

性能优化与工程落地建议

尽管LangChain提供了强大的抽象能力，但在生产环境中仍需注意以下几点：

1. 缓存中间结果避免重复计算

对于图像识别这类耗时操作，可引入缓存机制：

from langchain.globals import set_llm_cache from langchain_community.cache import InMemoryCache set_llm_cache(InMemoryCache())

注意：目前LangChain缓存主要针对LLM调用，自定义链需手动实现图像特征缓存。

2. 异步执行提升吞吐量

使用async版本链处理批量请求：

async def process_batch(image_paths): tasks = [overall_chain.ainvoke({"image_path": p}) for p in image_paths] results = await asyncio.gather(*tasks) return results

3. 错误处理与降级策略

try: result = vision_chain.run({"image_path": img_path}) except FileNotFoundError: fallback_result = {"category": "未知", "confidence": 0.0} except Exception as e: logging.error(f"图像识别失败: {e}") fallback_result = {"category": "系统错误", "confidence": 0.0}

4. 日志追踪与可观测性

集成LangChain内置回调系统，记录每一步执行时间与输入输出：

from langchain.callbacks import get_openai_callback with get_openai_callback() as cb: result = overall_chain.invoke({"image_path": "..."}) print(f"Total tokens: {cb.total_tokens}")

对比分析：LangChain vs 手动编排 vs 其他框架

| 维度 | LangChain方案 | 手动函数调用 | DSPy / LlamaIndex | |------|---------------|--------------|-------------------| | 开发效率 | ⭐⭐⭐⭐☆ | ⭐⭐ | ⭐⭐⭐ | | 可维护性 | ⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐☆ | | 动态编排能力 | ⭐⭐⭐☆ | ⭐ | ⭐⭐⭐⭐ | | 调试支持 | ⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐☆ | | 生产稳定性 | ⭐⭐☆ | ⭐⭐⭐⭐ | ⭐⭐⭐ |