Integrating AgentScope with AI security guardrails

Adds AI security guardrails to AgentScope by calling the Alibaba Cloud AI Security Guardrails service for real-time safety checks on large model inputs, outputs, and tool calls.

1. Core features

1.1 Detection point coverage

1.1.1 AgentScope 1.0 detection points

Reference: https://docs.agentscope.io/v1/building-blocks/hooking-functions

Hook name	Triggering time	Function
pre_reply	Before agent reply	User input detection
post_print	On streaming chunk output	Streaming incremental detection
post_reasoning	After LLM reasoning	Final output detection and violation replacement
toolkit middleware	Before tool execution	Tool input detection

Call sequence:

User input → [pre_reply input detection] → LLM streaming reasoning → [post_print incremental detection] → [post_reasoning final detection] → Tool call → [toolkit middleware tool detection] → Return to user

1.1.2 AgentScope 1.0 hook workflow diagram

Function of each interception point:

Interception point	Triggering time	Input	Interception method	Function
pre_reply	Before agent reply	User message text	Throws a ValueError	Detects user input and blocks violations immediately.
post_print	On each streaming chunk output	Accumulated text	Sets the is_blocked flag	Streaming incremental detection (submits every N characters).
post_reasoning	After LLM reasoning	Full output Msg	Replaces the output Msg	Final detection, buffer flushing, and violation replacement.
toolkit middleware	Before tool execution	Tool name and arguments	Yields a block message	Prevents dangerous tool calls.

1.1.3 AgentScope 2.0 detection points

Reference: https://docs.agentscope.io/v2/building-blocks/middleware

Method	Triggering time	Function
on_reply	Agent reply phase	User input detection and appends warnings to the output
on_reasoning	LLM reasoning phase	Streaming incremental output detection (submits every N characters)
on_acting	Tool call phase	Tool input detection

Call sequence:

User input → [on_reply input detection] → LLM streaming reasoning → [on_reasoning incremental detection] → Tool call → [on_acting tool detection] → Return to user

1.1.4 AgentScope 2.0 middleware workflow diagram

Function of each interception point:

Interception point	Location	Input	Output	Function
on_reply	Outermost layer	User message	Final reply	Detects user input / Appends a warning to the final output.
on_reasoning	Reasoning phase	tool_choice	Streaming text event + Msg	Detects model-generated text content (streaming).
on_model_call	Model API call	messages, tools	ChatResponse	Intercepts the prompt sent to the model and the model's raw response.
on_acting	Tool execution	tool_call	ToolResponse	Detects tool call arguments and prevents dangerous operations.
on_system_prompt	When building the prompt	Current prompt string	Transformed prompt	Dynamically modifies the system prompt.

Note: The current guardrail middleware implements the on_reply, on_reasoning, and on_acting interception points. on_model_call and on_system_prompt are extension points reserved by the AgentScope framework and are not used in this implementation.

1.2 Intelligent interception strategy

Decisions are based on the Suggestion and RiskLevel fields returned by the guardrail service:

Suggestion	RiskLevel	Action	Description
block	medium/high	Block	Immediately blocks medium- and high-risk content.
block	low	Warn	Issues a warning for low-risk content.
pass	-	Pass	Allows safe content to pass through.

1.3 Streaming detection

• threshold mode: Submits content for detection each time the buffer reaches a specified character count (default: 300).

• complete mode: Submits all content for detection after generation completes.

• Flushes remaining content at stream end.

• Supports overlap retention (default: 5 characters) to prevent semantic breaks at boundaries.

1.4 Large text chunking

• Automatically splits content exceeding 2000 characters into chunks.

• Calls the guardrail API concurrently with a configurable concurrency level.

• Aggregates detection results from all chunks for a final decision.

• Overlaps adjacent chunks by a configurable number of characters (default: 10) to maintain contextual coherence.

1.5 Overlap retention

To prevent semantic loss at chunk or streaming boundaries, configure the GUARDRAIL_OVERLAP_SIZE parameter:

1.5.1 Large text chunking example

Assume GUARDRAIL_MAX_CHUNK_SIZE=2000, GUARDRAIL_OVERLAP_SIZE=10:

Input text: 5500 characters
Chunk 1: text[0:2000]      # First chunk starts normally
Chunk 2: text[1990:3990]   # Overlaps with the previous chunk by 10 characters
Chunk 3: text[3980:5500]   # Overlaps with the previous chunk by 10 characters

1.5.2 Streaming detection example

Assume GUARDRAIL_STREAM_BUFFER_SIZE=300, GUARDRAIL_OVERLAP_SIZE=10:

1st submission: text[0:300] # Triggered after accumulating 300 characters
2nd submission: text[290:590]   # Overlaps with the end of the previous submission by 10 characters
3rd submission: text[580:880]   # Overlaps again by 10 characters
...

1.5.3 How it works

Scenario	No overlap	With overlap (overlap=10)
Chunk boundary	...illegal|manufacture... → Split into "illegal" and "manufacture", which are individually harmless.	...illegal manufacturing... The overlapping region maintains semantic integrity.
Stream boundary	1st check: "...how to" + 2nd check: "make a..." → Context is broken.	2nd check with overlap: "...how to make a..." → Context is coherent.

Recommended overlap: 3–20 (default: 10). Values that are too large increase redundant detection and degrade performance.

1.6 Tool call detection

Performs bidirectional safety detection on agent tool calls, including MCP services:

• Input validation: Concatenates the tool name and arguments into Tool: {name}\nArguments: {json} format for validation. Blocks execution if a violation is detected.

• Output detection: Extracts the text content of ToolResponse.content after execution and submits it for moderation. Replaces violating content with a blocking message.

1.7 Tracing parameters

Supported tracing parameters for troubleshooting and auditing:

Parameter	Description
request_id	Request ID
session_id	Session ID
user_id	User ID
trace_id	Tracing ID

2. Code integration and configuration

Base code and test cases for integrating AI security guardrails: agentscope_guardrails.zip

Version	Applicable framework	Integration mechanism	Directory
V1	AgentScope 1.0	Hook registration	v1/
V2	AgentScope 2.0	middleware	v2/

2.1 Project structure

├── guardrail_base.py           # Base classes: config management, service calls, response parsing
├── guardrail_service.py        # Alibaba Cloud AI security guardrails API calls
├── guardrail_text_checker.py   # Text checker (supports large text chunking)
├── guardrail_stream_checker.py # Stream checker (incremental submission)
├── v1/
│   └── guardrail_hook.py       # 1.0 Hook plugin
├── v2/
│   └── guardrail_middleware.py # 2.0 Middleware
└── test/
    ├── guardrail_test_v1.py    # 1.0 tests
    └── guardrail_test_v2.py    # 2.0 tests

2.2 Environment variables

Create the .env file:

# Alibaba Cloud AccessKey
GUARDRAIL_AK=your_access_key_id
GUARDRAIL_SK=your_access_key_secret

# Optional configuration
GUARDRAIL_SERVICE=agent_runtime_guard
GUARDRAIL_ENDPOINT=https://green-cip.cn-shanghai.aliyuncs.com
GUARDRAIL_MAX_CHUNK_SIZE=2000
GUARDRAIL_STREAM_BUFFER_SIZE=300
GUARDRAIL_OVERLAP_SIZE=10
GUARDRAIL_STREAM_CHECK_MODE=threshold
GUARDRAIL_MAX_CONCURRENT=5
GUARDRAIL_ENABLE_TRACING=true

2.3 V1: AgentScope 1.0 integration

Register guardrails with an existing Agent through the Hook mechanism. No Agent code changes are required.

import asyncio
import agentscope
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from v1.guardrail_hook import create_guardrail_hook
async def main():
    agentscope.init(project="GuardrailTest", name="Test")
    agent = ReActAgent(
        name="TestAgent",
        sys_prompt="You are a helpful assistant.",
        model=DashScopeChatModel(
            api_key="your-api-key",
            model_name="qwen-max",
            stream=True),
        formatter=DashScopeChatFormatter()
    )
    # Create and register the guardrail hook
    hook = create_guardrail_hook()
    hook.register_to_agent(agent)
    # Use the agent as usual; the guardrails are enabled automatically
    user_msg = Msg(name="user", content="Hello", role="user")
    try:
        response = await agent(user_msg)
        print(f"Agent: {response.content}")
    except ValueError as e:
        print(f"Blocked: {e.args[0].content}")
asyncio.run(main())

2.4 V2: AgentScope 2.0 integration

Inject guardrails into an Agent through middleware.

import asyncio
from agentscope.agent import Agent
from agentscope.credential import DashScopeCredential
from agentscope.formatter import DashScopeChatFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from v2.guardrail_middleware import create_guardrail_middleware
async def main():
    middleware = create_guardrail_middleware()
    model = DashScopeChatModel(
        credential=DashScopeCredential(api_key="your-api-key"),
        model="qwen-max",
        stream=True,
        formatter=DashScopeChatFormatter(),
    )
    agent = Agent(
        name="SafeAgent",
        system_prompt="You are a helpful assistant.",
        model=model,
        middlewares=[middleware],
    )
    user_msg = Msg(
        name="user",
        content=[TextBlock(type="text", text="Hello")],
        role="user",
    )
    response = await agent.reply(user_msg)
    print(f"Agent: {response.get_text_content()}")

asyncio.run(main())

2.5 Running tests

# V1 test
python test/guardrail_test_v1.py
# V2 test
python test/guardrail_test_v2.py

2.6 V1 vs. V2

Dimension	V1 (AgentScope 1.0)	V2 (AgentScope 2.0)
Extension mechanism	Hook (callback registration)	Middleware (lifecycle interception)
Input interception	Raises a ValueError exception	Yields a replacement message for a graceful stop
Stream processing	Indirectly handled by the post_print hook	on_reasoning directly consumes the event stream
Tool detection	toolkit.register_middleware()	on_acting unified processing
Integration method	Create the Agent, then register the Hook	Pass middlewares=[] during construction.

3. Notes

1. AccessKey security: Do not hard-code AccessKey credentials in your code. Use environment variables or a key management service.

2. Concurrency control: Default concurrency is 5. Setting this too high may cause request throttling.

3. Exception handling: By default, content passes if the guardrail service call fails (fail-open). Add monitoring and alerts for production environments.

4. Streaming detection: Default buffer size is 300 characters. Adjust based on your use case.

5. Overlap Retention: Default GUARDRAIL_OVERLAP_SIZE is 10 (recommended range: 3–20). Values too large increase redundant detection and hurt performance; values too small reduce boundary semantic preservation.