Ui0dZddlmZddlmZmZmZddlmZm Z m Z m Z ddl m Z ddlmZmZmZmZmZmZmZmZmZmZddlmZmZmZer ddlmZdd lm Z Gd d eZ!gd Z"y )z1PII detection and handling middleware for agents.) annotations) TYPE_CHECKINGAnyLiteral) AIMessage AnyMessage HumanMessage ToolMessage)override) PIIDetectionErrorPIIMatch RedactionRuleResolvedRedactionRuleapply_strategydetect_credit_card detect_email detect_ipdetect_mac_address detect_url)AgentMiddleware AgentState hook_config)Callable)RuntimeceZdZdZdddddd dfdZeddZdd Zed g e dd Z ed g  dd Z e ddZ ddZ xZS) PIIMiddlewarea Detect and handle Personally Identifiable Information (PII) in conversations. This middleware detects common PII types and applies configurable strategies to handle them. It can detect emails, credit cards, IP addresses, MAC addresses, and URLs in both user input and agent output. Built-in PII types: - `email`: Email addresses - `credit_card`: Credit card numbers (validated with Luhn algorithm) - `ip`: IP addresses (validated with stdlib) - `mac_address`: MAC addresses - `url`: URLs (both `http`/`https` and bare URLs) Strategies: - `block`: Raise an exception when PII is detected - `redact`: Replace PII with `[REDACTED_TYPE]` placeholders - `mask`: Partially mask PII (e.g., `****-****-****-1234` for credit card) - `hash`: Replace PII with deterministic hash (e.g., ``) Strategy Selection Guide: | Strategy | Preserves Identity? | Best For | | -------- | ------------------- | --------------------------------------- | | `block` | N/A | Avoid PII completely | | `redact` | No | General compliance, log sanitization | | `mask` | No | Human readability, customer service UIs | | `hash` | Yes (pseudonymous) | Analytics, debugging | Example: ```python from langchain.agents.middleware import PIIMiddleware from langchain.agents import create_agent # Redact all emails in user input agent = create_agent( "openai:gpt-5", middleware=[ PIIMiddleware("email", strategy="redact"), ], ) # Use different strategies for different PII types agent = create_agent( "openai:gpt-4o", middleware=[ PIIMiddleware("credit_card", strategy="mask"), PIIMiddleware("url", strategy="redact"), PIIMiddleware("ip", strategy="hash"), ], ) # Custom PII type with regex agent = create_agent( "openai:gpt-5", middleware=[ PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block"), ], ) ``` redactNTF)strategydetectorapply_to_inputapply_to_outputapply_to_tool_resultsc0t|||_||_||_t |||j |_|jj|_|jj|_ |jj|_ y)aInitialize the PII detection middleware. Args: pii_type: Type of PII to detect. Can be a built-in type (`email`, `credit_card`, `ip`, `mac_address`, `url`) or a custom type name. strategy: How to handle detected PII. Options: * `block`: Raise `PIIDetectionError` when PII is detected * `redact`: Replace with `[REDACTED_TYPE]` placeholders * `mask`: Partially mask PII (show last few characters) * `hash`: Replace with deterministic hash (format: ``) detector: Custom detector function or regex pattern. * If `Callable`: Function that takes content string and returns list of `PIIMatch` objects * If `str`: Regex pattern to match PII * If `None`: Uses built-in detector for the `pii_type` apply_to_input: Whether to check user messages before model call. apply_to_output: Whether to check AI messages after model call. apply_to_tool_results: Whether to check tool result messages after tool execution. Raises: ValueError: If `pii_type` is not built-in and no detector is provided. )pii_typerrN) super__init__r r!r"rresolve_resolved_ruler$rr)selfr$rrr r!r" __class__s l/var/www/html/eventheodds/airagagent/rag_env/lib/python3.12/site-packages/langchain/agents/middleware/pii.pyr&zPIIMiddleware.__init__^sR ,.%:"5B6  ')  ++44 ++44 ++44 cN|jjd|jdS)zName of the middleware.[])r*__name__r$)r)s r+namezPIIMiddleware.names&..))*!DMM?!<)r)stateruntimer: new_messages any_modified last_user_msg last_user_idxir3 new_contentr4updated_message last_ai_idxmsgtool_msgs r+ before_modelzPIIMiddleware.before_models(""4+E+E$H~      M M3x=1,b"5 hqk<8$,QKM$%M   (]}?T?Tm334'+'<' +(++*//3O 3BL/#'L  % %K3x=1,b"5 hqk95"#K  &{QH >,A"1+C!#{3#&'//$"%h&6&6"7/3/D/DW/M, W&$+6$/'{{!))1)>)> ++: Q'+ -,0  - -r,c.K|j||Sw)aAsync check user messages and tool results for PII before model invocation. Args: state: The current agent state. runtime: The langgraph runtime. Returns: Updated state with PII handled according to strategy, or `None` if no PII detected. Raises: PIIDetectionError: If PII is detected and strategy is `'block'`. )rPr)rDrEs r+ abefore_modelzPIIMiddleware.abefore_models&  00c|jsy|d}|syd}d}tt|dz ddD]}||}t|ts|}|}n||r |j syt |j }|j|\} } | syt | |j|j|j} t|} | | |<d| iS)aCheck AI messages for PII after model invocation. Args: state: The current agent state. runtime: The langgraph runtime. Returns: Updated state with PII handled according to strategy, or None if no PII detected. Raises: PIIDetectionError: If PII is detected and strategy is `'block'`. Nr:r;r<)r3r=r1 tool_calls) r!r@rArBrr3rCr6r=r1rVr?) r)rDrEr: last_ai_msgrMrJrNr3rKr4rLrFs r+ after_modelzPIIMiddleware.after_models&##$  s8}q("b1 A1+C#y)!      k9L9Lk))*#44W= W$~~!!"--  H~ $3 [!L))r,c.K|j||Sw)aAsync check AI messages for PII after model invocation. Args: state: The current agent state. runtime: The langgraph runtime. Returns: Updated state with PII handled according to strategy, or None if no PII detected. Raises: PIIDetectionError: If PII is detected and strategy is `'block'`. )rXrRs r+ aafter_modelzPIIMiddleware.aafter_modelTs$w//rT)r$zALiteral['email', 'credit_card', 'ip', 'mac_address', 'url'] | strrz*Literal['block', 'redact', 'mask', 'hash']rz,Callable[[str], list[PIIMatch]] | str | Noner boolr!r[r"r[returnNone)r\rC)r3rCr\ztuple[str, list[PIIMatch]])rDrrErr\zdict[str, Any] | None)r0 __module__ __qualname____doc__r&propertyr1r6rr rPrSrXrZ __classcell__)r*s@r+rrsP=J@HAE# %&+65T 65 = 65?656565 $65 65p=="eW% \\\  \&\|eW%111  1&1(9*9*9*  9*9*v000  0r,r)r rrrrrrN)#r` __future__rtypingrrrlangchain_core.messagesrrr r typing_extensionsr &langchain.agents.middleware._redactionr r rrrrrrrr!langchain.agents.middleware.typesrrrcollections.abcrlanggraph.runtimerr__all__r,r+rmsQ7"..TT&   WV()H0OH0V  r,