Astro Airflow调用instructor-xl报错，求LangChain嵌入服务调用方案

问题原因 #

你遇到的Negsignal.SIGKILL是因为Airflow任务进程内存不足，instructor-xl作为大尺寸嵌入模型，加载时会占用大量内存，被系统强制终止。将模型部署为独立服务是正确的解决思路，Airflow任务只需调用远程服务，无需本地加载模型。

步骤1：部署instructor-xl嵌入服务 #

用FastAPI搭建轻量嵌入服务示例（部署在内存充足的计算引擎上）：

from fastapi import FastAPI, Body
from instructor import INSTRUCTOR
import torch

app = FastAPI()

# 加载instructor-xl模型（建议GPU环境，显存需10GB以上）
model = INSTRUCTOR('hkunlp/instructor-xl')

@app.post("/embed")
def generate_embeddings(inputs: list[list[str]] = Body(...)):
    """
    输入格式：[[指令文本, 待嵌入文本], ...]
    示例：[["represent the document for retrieval", "这是需要嵌入的文档内容"]]
    """
    embeddings = model.encode(inputs)
    return {"embeddings": embeddings.tolist()}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8082)

启动服务后，可通过POST请求http://your-server-ip:8082/embed验证接口，传入上述格式的JSON即可获取嵌入向量。

步骤2：在LangChain中调用远程嵌入服务 #

自定义符合LangChain Embeddings规范的类，对接远程服务：

from langchain.embeddings.base import Embeddings
from typing import List
import requests

class RemoteInstructorEmbeddings(Embeddings):
    def __init__(self, server_url: str = "http://your-server-ip:8082/embed"):
        self.server_url = server_url

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        # 文档嵌入使用统一检索指令
        inputs = [["represent the document for retrieval", text] for text in texts]
        resp = requests.post(self.server_url, json=inputs)
        resp.raise_for_status()
        return resp.json()["embeddings"]

    def embed_query(self, text: str) -> List[float]:
        # 查询问题使用适配指令
        inputs = [["represent the question for retrieving relevant documents", text]]
        resp = requests.post(self.server_url, json=inputs)
        resp.raise_for_status()
        return resp.json()["embeddings"][0]

步骤3：结合向量数据库与LLM服务使用 #

将远程嵌入服务与你已有的LLM服务结合，完成检索问答流程：

from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import HuggingFaceTextGenInference

# 初始化远程嵌入服务
embeddings = RemoteInstructorEmbeddings(server_url="http://your-server-ip:8082/embed")
# 创建/加载向量数据库
db = Chroma.from_documents(your_documents, embeddings)
retriever = db.as_retriever()

# 初始化已有的LLM服务
llm = HuggingFaceTextGenInference(
    inference_server_url="http://localhost:8081/",
    max_new_tokens=1024,
    top_k=5,
    top_p=0.8,
    typical_p=0.95,
    temperature=0.5,
    repetition_penalty=1
)

# 构建检索问答链
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=retriever)

注意事项 #

• 部署模型的服务器需满足内存要求：GPU环境需10GB以上显存，CPU环境建议16GB以上内存。
• 若使用Hugging Face官方Inference Endpoints部署模型，可直接用LangChain的HuggingFaceInferenceAPI类，但需严格遵循instructor-xl的指令输入格式。

内容的提问来源于stack exchange，提问作者OctavianWR