> ## Documentation Index
> Fetch the complete documentation index at: https://docs.inferwall.com/llms.txt
> Use this file to discover all available pages before exploring further.
# InferenceWall Middleware for LangChain
> Add InferenceWall to LangChain using a wrapper function or InferenceWallCallback handler to scan user inputs and LLM outputs for threats automatically.
The LangChain integration gives you two ways to add InferenceWall scanning to any LangChain chat model: a simple **wrapper function** that handles the scan-call-scan loop inline, and an **`InferenceWallCallback` handler** that you call explicitly before and after the LLM invocation. Both approaches scan user input before it reaches the LLM and scan the LLM output before it reaches your user.
## Install
```bash theme={null}
pip install inferwall langchain langchain-openai
```
## Approach 1: Wrapper function
`guarded_chat` is a drop-in wrapper that scans the input, calls your LangChain model (or runs in demo mode without one), and scans the output — all in a single function call.
`guarded_chat` calls `inferwall.scan_input()` before touching the model. Blocked requests return a string message immediately; flagged requests log a warning and proceed.
```python theme={null}
import inferwall
input_scan = inferwall.scan_input(prompt)
if input_scan.decision == "block":
sigs = ", ".join(m["signature_id"] for m in input_scan.matches)
return f"[BLOCKED] Input rejected by security policy. Matched: {sigs}"
if input_scan.decision == "flag":
print(f"[WARNING] Input flagged (score={input_scan.score}), proceeding...")
```
Pass your chat model as the second argument. If you omit it, the function runs in demo mode with a simulated response.
```python theme={null}
from langchain_core.messages import HumanMessage
if chat_model is not None:
response = chat_model.invoke([HumanMessage(content=prompt)])
output_text = response.content
else:
# Demo mode — simulated LLM response
output_text = f"Here is my response to: {prompt}"
```
```python theme={null}
output_scan = inferwall.scan_output(output_text)
if output_scan.decision == "block":
sigs = ", ".join(m["signature_id"] for m in output_scan.matches)
return f"[BLOCKED] Output contained sensitive data. Matched: {sigs}"
if output_scan.decision == "flag":
print(f"[WARNING] Output flagged (score={output_scan.score})")
return output_text
```
## Approach 2: Callback handler
`InferenceWallCallback` gives you explicit control. Call `guard.on_input()` before invoking the model and `guard.on_output()` on the result. Both methods raise `ValueError` when the content is blocked, so your existing exception-handling code catches them naturally.
```python theme={null}
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
guard = InferenceWallCallback()
llm = ChatOpenAI(model="gpt-4o-mini")
```
Pass `block_on_flag=True` if you want flagged content treated as blocked:
```python theme={null}
guard = InferenceWallCallback(block_on_flag=True)
```
`on_input` returns the original text unchanged so you can chain the call. It raises `ValueError` if InferenceWall blocks the prompt.
```python theme={null}
prompt = "What is the weather?"
guard.on_input(prompt) # Raises if blocked
```
After the call, `guard.last_input_scan` holds the full `ScanResponse` for logging or auditing.
```python theme={null}
response = llm.invoke([HumanMessage(content=prompt)])
safe_output = guard.on_output(response.content) # Raises if blocked
```
`guard.last_output_scan` holds the output scan result.
## Complete example
```python theme={null}
from __future__ import annotations
import inferwall
# --- Option 1: Simple wrapper function ---
def guarded_chat(prompt: str, chat_model: object = None) -> str:
"""Scan input, call LLM, scan output, return safe response."""
# Scan user input
input_scan = inferwall.scan_input(prompt)
if input_scan.decision == "block":
sigs = ", ".join(m["signature_id"] for m in input_scan.matches)
return f"[BLOCKED] Input rejected by security policy. Matched: {sigs}"
if input_scan.decision == "flag":
print(f"[WARNING] Input flagged (score={input_scan.score}), proceeding...")
# Call LLM (replace with your actual LangChain call)
if chat_model is not None:
from langchain_core.messages import HumanMessage
response = chat_model.invoke([HumanMessage(content=prompt)])
output_text = response.content
else:
# Demo mode — simulated LLM response
output_text = f"Here is my response to: {prompt}"
# Scan LLM output
output_scan = inferwall.scan_output(output_text)
if output_scan.decision == "block":
sigs = ", ".join(m["signature_id"] for m in output_scan.matches)
return f"[BLOCKED] Output contained sensitive data. Matched: {sigs}"
if output_scan.decision == "flag":
print(f"[WARNING] Output flagged (score={output_scan.score})")
return output_text
# --- Option 2: LangChain callback handler ---
class InferenceWallCallback:
"""LangChain callback that scans inputs and outputs."""
def __init__(self, block_on_flag: bool = False) -> None:
self.block_on_flag = block_on_flag
self.last_input_scan: inferwall.ScanResponse | None = None
self.last_output_scan: inferwall.ScanResponse | None = None
def on_input(self, text: str) -> str:
"""Scan input. Raises ValueError if blocked."""
self.last_input_scan = inferwall.scan_input(text)
if self.last_input_scan.decision == "block":
raise ValueError(
f"Input blocked by InferenceWall "
f"(score={self.last_input_scan.score}, "
f"matches={[m['signature_id'] for m in self.last_input_scan.matches]})"
)
if self.block_on_flag and self.last_input_scan.decision == "flag":
raise ValueError(
f"Input flagged by InferenceWall "
f"(score={self.last_input_scan.score})"
)
return text
def on_output(self, text: str) -> str:
"""Scan output. Raises ValueError if blocked."""
self.last_output_scan = inferwall.scan_output(text)
if self.last_output_scan.decision == "block":
raise ValueError(
f"Output blocked by InferenceWall "
f"(score={self.last_output_scan.score}, "
f"matches={[m['signature_id'] for m in self.last_output_scan.matches]})"
)
return text
```
You can access `guard.last_input_scan` and `guard.last_output_scan` after each call to retrieve the full `ScanResponse` for logging, metrics, or tracing.
## Agent compatibility
The callback handler approach works with LangChain agents: call `guard.on_input()` before invoking the agent and wrap tool outputs with `guard.on_output()` before passing them back into the agent loop. The wrapper function is better suited to simple chain or single-turn invocations.
## Related
* [How InferenceWall works](/concepts/how-it-works)
* [Policies and thresholds](/concepts/policies)