refactor(messageProcessor): integrate PII redaction in message processing

- Updated the MessageProcessor to utilize the new redactTextUsingPII function for redacting sensitive information from messages.
- Changed buildMemory method to be asynchronous, ensuring proper handling of redaction before processing messages.
- Enhanced type definitions for callback parameters to improve code clarity and maintainability.

Author: nicky-ru

packages/core/src/PII.ts

@@ -1,4 +1,5 @@
 import { pipeline } from "@huggingface/transformers";
+import elizaLogger from "./logger";
 
 type RedactionEntity = {
     type: string;
@@ -394,3 +395,38 @@ export class PII {
         }
     }
 }
+
+let piiInstance: PII | null = null;
+let piiInitPromise: Promise<void> | null = null;
+
+async function ensurePIIInitialized(): Promise<PII | null> {
+    if (piiInstance) return Promise.resolve(piiInstance);
+    if (!piiInitPromise) {
+        piiInitPromise = PII.create()
+            .then((pii) => {
+                piiInstance = pii;
+            })
+            .catch((error) => {
+                elizaLogger.error(
+                    "Failed to initialize PII for message redaction:",
+                    error
+                );
+                piiInstance = null;
+            });
+    }
+    return piiInitPromise.then(() => piiInstance);
+}
+
+export async function redactTextUsingPII(text: string): Promise<string> {
+    if (process.env.PII_REDACTION !== "true") return text;
+    if (typeof text !== "string" || text.length === 0) return text;
+    const pii = await ensurePIIInitialized();
+    if (!pii) return text;
+    try {
+        const result = await pii.redact(text);
+        return result?.redactedText ?? text;
+    } catch (error) {
+        elizaLogger.error("PII redaction failed in message processor:", error);
+        return text;
+    }
+}

packages/core/src/messageProcessor.ts

@@ -14,6 +14,7 @@ import {
 import { composeContext } from "./context";
 import { generateMessageResponse } from "./generation";
 import { InteractionLogger, AgentClient } from "./interactionLogger";
+import { redactTextUsingPII } from "./PII";
 
 export interface ReceivedMessage {
     rawMessageId: string;
@@ -54,7 +55,7 @@ export class MessageProcessor {
             message.source
         );
 
-        this.messageToProcess = this.buildMemory(message);
+        this.messageToProcess = await this.buildMemory(message);
         await this.saveMemory(this.messageToProcess);
         this.state = await this.runtime.composeState(this.messageToProcess);
 
@@ -73,7 +74,7 @@ export class MessageProcessor {
 
             const callbackWithMemorySaving = async (
                 content: Content,
-                files: any[]
+                files: Array<{ attachment: string; name: string }>
             ) => {
                 const memories = await callback(content, files);
                 await this.saveMemories(memories);
@@ -157,9 +158,11 @@ export class MessageProcessor {
         });
     }
 
-    private buildMemory(message: ReceivedMessage): Memory {
+    private async buildMemory(message: ReceivedMessage): Promise<Memory> {
+        const redactedText = await redactTextUsingPII(message.text);
+
         const content: Content = {
-            text: message.text,
+            text: redactedText,
             attachments: message.attachments,
             source: message.source,
             inReplyTo: message.inReplyTo,