在构建聊天应用时，确保敏感数据的安全处理至关重要，特别是个人身份信息（PII）。PII 可以直接或间接关联到个人，因此必须通过阻止此类数据传输到语言模型来保护用户隐私。

​
PII 示例

考虑以下文本，其中 PII 已被高亮显示

你好，我的名字是 John，我住在 纽约。我的信用卡号是 3782-8224-6310-005，我的电话号码是 (212) 688-5500。

以下是匿名化版本

你好，我的名字是 <PERSON>，我住在 <LOCATION>。我的信用卡号是 <CREDIT_CARD>，我的电话号码是 <PHONE_NUMBER>。

​
分析和匿名化数据

集成 Microsoft Presidio 以在你的 Chainlit 应用中实现强大的数据清洗。

代码示例
import chainlit as cl

@cl.on_message
async def main(message: cl.Message):
    # Notice that the message is passed as is
    response = await cl.Message(
        content=f"Received: {message.content}",
    ).send()

在继续之前，请确保已安装进行 PII 分析和匿名化所需的 Python 包。在你的终端中运行以下命令进行安装

pip install presidio-analyzer presidio-anonymizer spacy
python -m spacy download en_core_web_lg

创建一个异步上下文管理器，利用 Presidio Analyzer 检查传入文本中是否存在任何 PII。可以将此上下文管理器包含在你的主函数中，以便在处理消息之前对其进行仔细检查。当检测到 PII 时，你应该向用户提供继续或取消操作的选项。使用 Chainlit 的消息系统来实现此目的。

代码示例
from presidio_analyzer import AnalyzerEngine
from contextlib import asynccontextmanager

analyzer = AnalyzerEngine()

@asynccontextmanager
async def check_text(text: str):
  pii_results = analyzer.analyze(text=text, language="en")

  if pii_results:
    response = await cl.AskActionMessage(
      content="PII detected",
      actions=[
        cl.Action(name="continue", payload={"value": "continue"}, label="✅ Continue"),
        cl.Action(name="cancel", payload={"value": "continue"}, label="❌ Cancel"),
      ],
    ).send()

    if response is None or response.get("payload").get("value") == "cancel":
      raise InterruptedError

  yield

# ...

@cl.on_message
async def main(message: cl.Message):
  async with check_text(message.content):
    # This block is only executed when the user press "Continue"
    response = await cl.Message(
        content=f"Received: {message.content}",
    ).send()

如果你的应用需要匿名化 PII，Presidio 也可以做到。修改 check_text 上下文管理器，以便在检测到 PII 时返回匿名化文本。

代码示例
from presidio_anonymizer import AnonymizerEngine

anonymizer = AnonymizerEngine()

@asynccontextmanager
async def check_text(text: str):
  pii_results = analyzer.analyze(text=text, language="en")

  if pii_results:
    response = await cl.AskActionMessage(
      content="PII detected",
      actions=[
        cl.Action(name="continue", payload={"value": "continue"}, label="✅ Continue"),
        cl.Action(name="cancel", payload={"value": "continue"}, label="❌ Cancel"),
      ],
    ).send()

    if response is None or response.get("payload").get("value") == "cancel":
      raise InterruptedError

    yield anonymizer.anonymize(
      text=text,
      analyzer_results=pii_results,
    ).text
  else:
    yield text

# ...

@cl.on_message
async def main(message: cl.Message):
  async with check_text(message.content) as anonymized_message:
    response = await llm_chain.arun(
      anonymized_message
      callbacks=[cl.AsyncLangchainCallbackHandler()]
    )