在阿里云与 Azure Functions 间构建基于 mTLS 的 LangChain 服务通信及 CAP 权衡实践

分布式架构

文章字数: 4.5k

阅读时长: 18 分

一个生产级系统面临的技术决策，往往不是单一维度的“好”与“坏”，而是在多个相互制约的目标中寻找一个动态平衡点。当前我们面临的挑战是：一个基于 LangChain 的文档分析代理，其核心数据处理与向量化逻辑，因数据主权与存算成本考虑，部署在阿里云的 ECS 实例上。而其对外提供服务的推理端点，则希望利用 Azure Functions 的弹性伸缩和低运维成本特性。

这立刻引出了两个核心问题：

安全（Security）: 部署在阿里云 VPC 内的后端服务，如何跨越公网，以零信任的方式与 Azure Functions 端点进行通信？简单的 API Key 认证在这种场景下是脆弱的。
可靠性（Reliability）: 这是一个跨云的分布式系统，网络分区是必然会发生的事件。根据 CAP 定理，我们必须在一致性（Consistency）和可用性（Availability）之间做出明确取舍。对于一个处理敏感信息的 AI 应用，这个决策将直接影响其行为模式和用户信任。

架构决策：通信协议的权衡

在解决跨云通信安全问题上，我们评估了三种主流方案。

方案 A：API Key + IP 白名单

这是最直接的方案。Azure Functions 提供基于密钥的授权，我们可以在阿里云的客户端代码中携带此密钥，并在 Azure Functions 的防火墙中设置阿里云 ECS 的出口 IP 白名单。

优势: 实现简单，几乎没有额外的开发成本。
劣势:
- 弱身份验证: API Key 本质上是“共享密钥”，一旦泄露，冒充攻击将变得轻而易举。它只能验证“请求者知道密钥”，但无法验证“请求者是谁”。
- 静态与脆弱: 在动态云环境中，IP 地址可能会改变。依赖 IP 白名单会带来运维负担，且无法防御来自同一信任网络内部的恶意流量。
- 单向认证: 它只验证了客户端对服务端的访问权限，服务端对客户端的身份一无所知。这不符合零信任（Zero Trust）原则。

在真实项目中，这种方案仅适用于内部低安全级别的服务调用。对于跨越公网、处理敏感数据的场景，其安全保障是完全不足的。

方案 B：构建跨云 VPN 隧道

我们可以通过在阿里云 VPC 和 Azure VNet 之间建立站点到站点的 VPN 隧道，将两个云环境从网络层面打通，使得 ECS 可以像访问内网服务一样调用 Azure Functions。

优势: 提供了网络层的整体加密和隔离，对上层应用透明。
劣势:
- 成本与复杂性: 无论是阿里云的 CEN 还是 Azure 的 VPN Gateway，都涉及显著的持续费用和复杂的配置、监控与维护工作。
- 过度设计: 我们的需求仅仅是两个特定服务间的点对点安全通信，为此建立一个完整的网络隧道，属于“杀鸡用牛刀”。
- 性能瓶颈: 所有的流量都必须经过 VPN 网关，这可能成为性能瓶颈和单点故障。

最终选择：双向 TLS 认证 (mTLS)

mTLS 在标准 TLS 的基础上，要求客户端也提供自己的证书以供服务端校验。这样，通信双方都能确认对方的真实身份，实现了双向的、基于密码学的强认证。

优势:
- 强身份认证: 基于非对称加密的证书体系，提供了远超共享密钥的安全性。
- 应用层安全: 安全性与应用层绑定，不依赖底层网络拓扑，灵活且适应性强。
- 精细化控制: 我们可以为每个客户端服务签发独立的证书，实现细粒度的访问控制和吊销。
- 成本效益: 除了证书管理的开销，它不引入额外的基础设施成本。

这个方案完美契合我们的需求：为跨公网的服务调用建立一个加密的、双向认证的、零信任的安全通道。接下来的核心任务，就是具体实现它。

核心实现：构建 mTLS 通信链路

实现 mTLS 的核心在于证书管理和两端的正确配置。我们将自建一个简单的 CA (Certificate Authority) 来签发证书。在生产环境中，一个常见的错误是图方便而使用自签名证书，这会丧失证书链验证的意义，应当使用专用的 CA 体系或 Vault 等工具管理。

1. 证书体系的构建

我们将使用 OpenSSL 创建一个根 CA，并用它来签发服务端（Azure Functions）和客户端（Alibaba Cloud ECS）的证书。

# --- 1. 创建根CA ---
# 生成CA私钥
openssl genrsa -out rootCA.key 4096

# 生成CA根证书 (有效期10年)
openssl req -x509 -new -nodes -key rootCA.key -sha256 -days 3650 -out rootCA.pem \
    -subj "/C=CN/ST=Beijing/L=Beijing/O=MyOrg/OU=CA/CN=my-internal-ca"

# --- 2. 为 Azure Functions (服务端) 签发证书 ---
# 生成服务端私钥
openssl genrsa -out server.key 2048

# 生成服务端证书签名请求 (CSR)
# 这里的CN必须是Azure Function App的域名, 例如: my-langchain-func.azurewebsites.net
openssl req -new -key server.key -out server.csr \
    -subj "/C=CN/ST=Beijing/L=Beijing/O=MyOrg/OU=Server/CN=my-langchain-func.azurewebsites.net"

# 使用CA签发服务端证书 (有效期1年)
openssl x509 -req -in server.csr -CA rootCA.pem -CAkey rootCA.key -CAcreateserial -out server.crt -days 365 -sha256

# --- 3. 为 Alibaba Cloud ECS (客户端) 签发证书 ---
# 生成客户端私钥
openssl genrsa -out client.key 2048

# 生成客户端证书签名请求 (CSR)
# CN可以用来标识客户端身份, 例如: alicloud-ecs-retriever-service
openssl req -new -key client.key -out client.csr \
    -subj "/C=CN/ST=Beijing/L=Beijing/O=MyOrg/OU=Client/CN=alicloud-ecs-retriever-service"

# 使用CA签发客户端证书
openssl x509 -req -in client.csr -CA rootCA.pem -CAkey rootCA.key -CAcreateserial -out client.crt -days 365 -sha256

执行完毕后，我们得到了以下关键文件：

rootCA.pem: CA 根证书，通信双方都需要用它来验证对方证书的合法性。
server.key, server.crt: 服务端私钥和证书。
client.key, client.crt: 客户端私钥和证书。

2. 服务端实现：Azure Functions 配置

Azure Functions 对客户端证书提供了内建支持。我们需要进行两步配置：

首先，在 Function App 的“TLS/SSL 设置” -> “专用密钥证书 (.pfx)”中，上传包含服务端证书和私钥的 PFX 文件。然后，在“传入客户端证书”中，启用“允许传入客户端证书”并选择“允许但不要求”。我们将在代码中强制校验。

其次，修改 host.json 文件，配置证书验证。

// host.json
{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensions": {
    "http": {
      "routePrefix": ""
    }
  },
  "functionAppLogs": {
    "logLevel": {
      "default": "Information"
    }
  },
  // 核心配置：开启客户端证书转发
  "managedDependency": {
    "enabled": true
  },
  "customHandler": null,
  "healthMonitor": {
    "enabled": true
  },
  "http": {
    // 这个设置将客户端证书的头信息转发给我们的函数代码
    "forwarding-gateway": {
      "x-forwarded-client-cert": {
        "header-name": "X-ARR-ClientCert"
      }
    }
  }
}

现在，我们可以在函数代码中编写校验逻辑。这里的关键是，Azure App Service 会将客户端证书通过 X-ARR-ClientCert HTTP 头传递给我们的函数。我们需要解析这个头，并验证证书的指纹（Thumbprint）或主题（Subject）。

# LlmInferenceFunction/__init__.py

import logging
import azure.functions as func
import json
import os
from cryptography import x509
from cryptography.hazmat.backends import default_backend

# 从应用设置中读取允许的客户端证书指纹列表
# 这是一个更安全的实践，而不是硬编码在代码里
ALLOWED_CLIENT_THUMBPRINTS = os.environ.get("ALLOWED_CLIENT_THUMBPRINTS", "").split(',')

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    # --- mTLS 验证逻辑 ---
    client_cert_header = req.headers.get('X-ARR-ClientCert')
    if not client_cert_header:
        logging.error("Client certificate missing.")
        return func.HttpResponse("Forbidden: Client certificate required.", status_code=403)

    try:
        # PEM 格式的证书数据需要补全首尾
        cert_pem = f"-----BEGIN CERTIFICATE-----\n{client_cert_header}\n-----END CERTIFICATE-----"
        cert_data = cert_pem.encode('utf-8')
        
        # 使用 cryptography 库解析证书
        cert = x509.load_pem_x509_certificate(cert_data, default_backend())
        
        # 获取证书指纹 (SHA1 Thumbprint)
        thumbprint = cert.fingerprint(cert.signature_hash_algorithm).hex().upper()
        logging.info(f"Received client certificate with thumbprint: {thumbprint}")

        # 校验指纹是否在白名单中
        if thumbprint not in ALLOWED_CLIENT_THUMBPRINTS:
            logging.warning(f"Unauthorized client certificate thumbprint: {thumbprint}")
            return func.HttpResponse("Forbidden: Invalid client certificate.", status_code=403)

    except Exception as e:
        logging.error(f"Error processing client certificate: {e}")
        return func.HttpResponse("Forbidden: Could not process client certificate.", status_code=403)
    
    logging.info("Client certificate validation successful.")
    
    # --- 业务逻辑 ---
    try:
        req_body = req.get_json()
        # 假设这里是调用LangChain的LLM部分
        query = req_body.get('query')
        context = req_body.get('context')
        
        # 伪代码：实际调用LLM进行推理
        # response_text = invoke_llm_chain(query, context)
        response_text = f"LLM response for '{query}' with provided context."

        return func.HttpResponse(
            json.dumps({"response": response_text}),
            mimetype="application/json",
            status_code=200
        )
    except ValueError:
        return func.HttpResponse("Invalid request body.", status_code=400)
    except Exception as e:
        logging.error(f"Function execution error: {e}")
        return func.HttpResponse("Internal server error.", status_code=500)

在 Azure Function App 的“配置” -> “应用程序设置”中，我们添加一个名为 ALLOWED_CLIENT_THUMBPRINTS 的键，其值为我们之前生成的 client.crt 的 SHA1 指纹。这样，只有持有正确客户端证书的服务才能调用此函数。

3. 客户端实现：阿里云 ECS 上的 LangChain 服务

在阿里云 ECS 上，我们有一个 Python 服务，它集成了 LangChain 的文档加载和向量检索部分。当需要语言模型进行最终生成时，它会调用 Azure Functions。

# alicloud_retriever_service.py

import requests
import os
import logging
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# --- 配置 ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 从环境变量或配置文件加载路径和URL，避免硬编码
CLIENT_CERT_PATH = os.environ.get("CLIENT_CERT_PATH", "client.crt")
CLIENT_KEY_PATH = os.environ.get("CLIENT_KEY_PATH", "client.key")
# 这是CA根证书，用于验证服务端证书的合法性
CA_CERT_PATH = os.environ.get("CA_CERT_PATH", "rootCA.pem") 
AZURE_FUNCTION_URL = os.environ.get("AZURE_FUNCTION_URL") # 例如: https://my-langchain-func.azurewebsites.net/api/LlmInferenceFunction

# 设置 requests session 以复用连接和mTLS配置
# 在生产代码中，session应该是长生命周期的对象
session = requests.Session()
session.cert = (CLIENT_CERT_PATH, CLIENT_KEY_PATH)
session.verify = CA_CERT_PATH # 指定CA证书用于验证服务端

# --- 模拟 LangChain 组件 ---

def local_document_retriever(query: str) -> str:
    """
    一个伪实现，模拟从本地向量数据库检索相关文档片段。
    在真实项目中，这里会连接到 Faiss, Milvus, or Elasticsearch。
    """
    logging.info(f"Retrieving documents for query: '{query}'")
    # 模拟检索结果
    retrieved_context = "CAP 定理指出，任何一个分布式系统最多只能同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance）这三项中的两项。"
    return retrieved_context


def invoke_azure_llm(input_dict: dict) -> str:
    """
    通过 mTLS 调用 Azure Function 上的 LLM 推理服务。
    这个函数是 LangChain 链中的一个关键环节。
    """
    query = input_dict.get("query")
    context = input_dict.get("context")
    
    payload = {
        "query": query,
        "context": context
    }
    
    try:
        # 设定合理的超时，这是实现 CAP 权衡的第一步
        response = session.post(AZURE_FUNCTION_URL, json=payload, timeout=5.0)
        
        # 严格的错误处理
        response.raise_for_status() # 如果状态码是 4xx 或 5xx，将抛出异常
        
        result = response.json()
        logging.info("Successfully received response from Azure Function.")
        return result.get("response", "Error: malformed response.")

    except requests.exceptions.Timeout:
        logging.error("Request to Azure Function timed out.")
        # 这里是 CAP 决策点: 超时后是失败还是降级？
        raise ConnectionError("LLM service is unavailable (timeout).")
    except requests.exceptions.SSLError as e:
        logging.error(f"mTLS handshake failed: {e}")
        # SSL错误通常是配置问题，直接失败是合理的
        raise ConnectionError(f"mTLS handshake failed: {e}")
    except requests.exceptions.RequestException as e:
        logging.error(f"Request to Azure Function failed: {e}")
        raise ConnectionError(f"LLM service is unavailable (request failed).")


# --- 构建 LangChain 链 ---
template = """
Based on the following context, answer the user's question.
Context: {context}
Question: {query}
Answer:
"""
prompt = ChatPromptTemplate.from_template(template)

# 使用 RunnableLambda 将我们的函数包装成 LangChain 组件
retriever_runnable = RunnableLambda(local_document_retriever)
azure_llm_runnable = RunnableLambda(invoke_azure_llm)

# 组装完整的 RAG 链
# RunnablePassthrough.assign 允许我们将检索器的输出（context）和原始问题（query）一起传递给下一个环节
chain = (
    {"context": retriever_runnable, "query": RunnablePassthrough()}
    | azure_llm_runnable
)


# --- 执行 ---
if __name__ == "__main__":
    if not all([CLIENT_CERT_PATH, CLIENT_KEY_PATH, CA_CERT_PATH, AZURE_FUNCTION_URL]):
        logging.critical("Missing required environment variables for mTLS configuration.")
    else:
        try:
            user_query = "Explain CAP theorem"
            final_response = chain.invoke(user_query)
            print("--- Final Response ---")
            print(final_response)
        except ConnectionError as e:
            print(f"\n--- Execution Failed ---")
            print(f"Could not complete the chain due to a communication failure: {e}")
            # 在这里，我们可以决定是简单地向用户报告错误，还是触发备用逻辑

现在，我们有了一个完整的、基于 mTLS 的跨云服务调用链路。阿里云的服务可以安全地将任务分派给 Azure Functions 上的推理端点。

架构的韧性：CAP 定理下的实践抉择

我们的系统横跨两个云厂商，通过公网连接。这意味着网络分区（P）不是一个“如果”会发生的问题，而是一个“何时”会发生的问题。根据 CAP 定理，我们必须在一致性（C）和可用性（A）之间做出选择。

graph TD
    subgraph Alibaba Cloud ECS
        A[User Query] --> B{LangChain Agent};
        B --> C[Retrieve Docs];
    end
    
    subgraph Azure Functions
        E[Invoke LLM]
    end

    C -- mTLS Request --> D{Network};
    D -- Potentially Partitioned --> E;

    subgraph Decision Point
        F{Handle Network Failure?};
        D -- Failure Signal --> F;
        F -- Option 1: CP --> G[Fail Fast & Error];
        F -- Option 2: AP --> H[Fallback Logic / Stale Data];
    end

    E --> I[LLM Response];
    I --> D;
    B --> F;

场景 1：选择一致性（CP - Consistency/Partition Tolerance）

对于金融、医疗等领域的问答系统，一个错误或基于过时信息的答案可能导致严重后果。在这种情况下，一致性远比可用性重要。

实现策略:
1. 严格的超时与重试: 在 invoke_azure_llm 函数中，设置一个较短的超时时间（如3-5秒）。
2. 有限的重试: 实现一个带指数退避的重试机制，但总次数不超过2次。过多的重试在网络持续中断时是无意义的，反而会耗尽客户端资源。
3. 快速失败 (Fail Fast): 如果重试后仍然失败，函数必须立即向上层调用栈抛出明确的异常（如 ConnectionError）。
4. 无降级逻辑: LangChain 链在接收到这个异常后，整个调用过程终止，并向最终用户返回一个明确的错误信息，例如：“服务当前不可用，请稍后重试”。
代码体现:
我们在 alicloud_retriever_service.py 的 invoke_azure_llm 函数中已经通过 response.raise_for_status() 和 try...except 块实现了这个模式。任何网络层或应用层的失败（超时、HTTP 5xx、403 Forbidden等）都会被捕获并转化为一个 ConnectionError，从而中断链的执行。

场景 2：选择可用性（AP - Availability/Partition Tolerance）

对于一个非关键的内部知识库或通用聊天机器人，保证服务时刻可用，即使用户得到的回应不完美，也可能比直接返回错误要好。

实现策略:
1. 降级逻辑 (Fallback): 在捕获到 ConnectionError 时，不立即抛出异常，而是调用一个备用逻辑。
2. 备用逻辑选项:
  - 使用缓存: 如果之前对类似查询有成功的 LLM 响应，可以返回一个缓存的、可能略微过时的答案，并附带一个提示（“此信息可能不是最新的”）。
  - 使用简化模型: 调用一个更小的、甚至可以在本地部署的语言模型（如 a small version of BERT for classification）来给出一个基础的、非生成式的回答。
  - 返回检索内容: 最简单的降级，直接返回 local_document_retriever 检索到的原始文本片段，让用户自己阅读。
代码修改示例:

# alicloud_retriever_service.py (AP 版本修改)

def invoke_azure_llm_ap_version(input_dict: dict) -> str:
    """
    通过 mTLS 调用 Azure Function，但在失败时提供降级逻辑。
    """
    query = input_dict.get("query")
    context = input_dict.get("context")
    
    try:
        # ... 省略与CP版本相同的请求代码 ...
        response = session.post(AZURE_FUNCTION_URL, json={"query": query, "context": context}, timeout=5.0)
        response.raise_for_status()
        result = response.json()
        return result.get("response", "Error: malformed response.")

    except requests.exceptions.RequestException as e:
        logging.warning(f"Request to Azure Function failed: {e}. Executing fallback logic.")
        # AP 模式: 提供降级响应而不是失败
        return f"The LLM service is currently unavailable. Based on our documents, here is the relevant information:\n\n---\n{context}\n---"

# 在构建链时使用这个新函数
azure_llm_runnable_ap = RunnableLambda(invoke_azure_llm_ap_version)
chain_ap = (
    {"context": retriever_runnable, "query": RunnablePassthrough()}
    | azure_llm_runnable_ap
)

# ... main函数中调用 chain_ap.invoke(...)

这个决策不是一次性的，它应该基于业务场景，甚至可以动态调整。一个常见的错误是在设计分布式系统时不明确考虑分区容错，导致系统在网络波动时出现不可预测的行为。

局限性与未来展望

当前架构虽然解决了核心的安全和通信问题，但在生产环境中仍有几个方面需要持续关注：

证书生命周期管理: 手动生成和分发证书是不可扩展且不安全的。在生产环境中，必须引入自动化工具，如 HashiCorp Vault 或云厂商提供的密钥管理服务（如 Azure Key Vault, Alibaba Cloud KMS/Secrets Manager），来自动化证书的签发、轮换和吊销。
性能开销: mTLS 的握手过程，特别是初次连接，会带来额外的延迟。对于需要极低延迟的应用，需要评估这个开销，并利用 HTTP Keep-Alive 和 TLS Session Resumption 等技术来优化。
可观测性: 跨云链路的监控是一个挑战。我们需要整合两边的日志和指标，建立统一的仪表盘来监控 mTLS 握手成功率、请求延迟、错误率等关键 SLI (Service Level Indicators)，以便在问题发生时能快速定位是在客户端、服务端还是中间网络。
服务发现: 当前架构中，Azure Function 的 URL 是硬编码或通过环境变量配置的。当服务扩展到多个区域或有更复杂的部署拓扑时，需要引入服务发现机制。

这个架构的本质是在异构和地理分布的计算单元之间，建立起一个安全、可靠且行为可预测的通信模型。mTLS 提供了安全基础，而对 CAP 定理的深刻理解与应用，则决定了系统在逆境中的表现。

Alibaba Cloud mTLS Azure Functions LangChain CAP Theorem

构建一个基于 Clojure、Node.js 与 OpenSearch 的异构事件溯源状态管理引擎

2023-10-27 架构与设计

Node.js Clojure OpenSearch 状态管理事件溯源 CQRS

构建服务于 Gatsby 静态站点与原生 iOS 客户端的 Quarkus 统一 API

2023-10-27 后端架构

iOS 开发 Gatsby Quarkus Scrum API 设计