Integrate with AgentScope

This recipe is for developers using AgentScope as the runtime for their AI agents. It demonstrates how to connect Knowledge Studio to an AgentScope Agent via a custom middleware, serving as the agent's knowledge memory. Fully code-driven, no console operations required. Code verified with AgentScope 2.0.4 + Knowledge Studio production environment.

Principle: Knowledge Base as Agent Memory

LLMs are inherently stateless — each inference is independent. AgentScope uses a middleware mechanism to inject external capabilities into agents (RAG retrieval, long-term memory, tracing, etc.) without modifying the agent's core logic. AgentScope ships with a built-in RAGMiddleware that works with KnowledgeBase objects (wrapping a local embedding model and vector store) and injects retrieval results before reasoning. Knowledge Studio, however, is a managed service — embedding, vector storage, and retrieval all run server-side, eliminating the need for a local vector store. This recipe implements a custom KnowledgeStudioRAGMiddleware that calls the Knowledge Studio knowledge search API directly, integrating retrieval results into the AgentScope Agent in two modes:

Mode	When	Implementation	AgentScope Middleware Hook
static	Before first reasoning in each reply	Use user message as query, inject results into system prompt	`on_system_prompt`
agentic	Model decides	Expose `search_knowledge` tool, agent calls autonomously	`list_tools`

AgentScope middleware has 6 hook positions. This recipe uses two: on_system_prompt (Transformer type, serial pipeline modifying system prompt) and list_tools (Tool source type, declaring tools provided by the middleware). See AgentScope Middleware Docs.

Prerequisites

Install Dependencies

uv pip install "agentscope[full]" aiohttp

AgentScope 2.0 requires Python 3.11+. Recommended to install with uv. See AgentScope Quick Start.

Required Parameters

Parameter	How to Obtain	Example
DashScope API Key	Create at Settings → API Key	`sk-xxxxxxxx`
Workspace ID	Business space ID from the console URL	`llm-xxxxxxxx`
Knowledge Search Service ID	Create and publish at Knowledge Service → Knowledge Search	`aid-xxxxxxxx`

Step 1: Build Knowledge Base via API

Knowledge base setup (file upload → create index → import) is fully API-driven. The following wraps it into a reusable function. Full API details: Create Index and Register File.

import os
import time
import hashlib
import requests

BASE_URL = "https://{workspaceId}.cn-beijing.maas.aliyuncs.com"
API_KEY = os.environ["DASHSCOPE_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


def upload_file(file_path: str, category: str = "cate_demo") -> str:
    """Upload file to data center, return fileId."""
    file_size = str(os.path.getsize(file_path))
    file_name = os.path.basename(file_path)
    with open(file_path, "rb") as f:
        content_md5 = hashlib.md5(f.read()).hexdigest()

    # 1. Request upload lease
    resp = requests.post(
        f"{BASE_URL}/api/v1/connector/dash/applyFileUploadLease",
        headers=HEADERS,
        json={"category": category, "fileName": file_name,
              "sizeBytes": file_size, "contentMd5": content_md5})
    lease = resp.json()["data"]

    # 2. Upload to OSS
    with open(file_path, "rb") as f:
        requests.put(lease["param"]["url"],
                      headers=lease["param"]["headers"], data=f)

    # 3. Register file
    resp = requests.post(
        f"{BASE_URL}/api/v1/connector/dash/addFile",
        headers=HEADERS,
        json={"leaseId": lease["leaseId"], "category": category,
              "categoryType": "UNSTRUCTURED", "parser": "AUTO_SELECT"})
    return resp.json()["data"]["fileId"]


def create_kb_and_wait(file_ids: list[str], kb_name: str = "my-kb") -> str:
    """Create knowledge base and import files, poll until complete."""
    resp = requests.post(
        f"{BASE_URL}/api/v1/indices/rag/index/create_v2",
        headers=HEADERS,
        json={"name": kb_name, "structureType": "unstructured",
              "sinkType": "DEFAULT", "sourceType": "DATA_CENTER_FILE",
              "embeddingModelName": "text-embedding-v4", "chunkSize": 600,
              "docIds": file_ids,
              "dataSources": [{"sourceType": "DATA_CENTER_FILE"}]})
    data = resp.json()["data"]
    kb_id, job_id = data["pipelineId"], data["ingestionId"]

    while True:
        resp = requests.get(
            f"{BASE_URL}/api/v1/indices/rag/index_job/status",
            headers=HEADERS,
            params={"index_id": kb_id, "job_id": job_id})
        status = resp.json()["data"]["ingestion_status"]
        if status == "COMPLETED":
            break
        elif status in ("FAILED", "CANCELLED"):
            raise RuntimeError(f"Import failed: {status}")
        time.sleep(3)
    return kb_id


# Build knowledge base in one call
file_id = upload_file("product-guide.md")
kb_id = create_kb_and_wait([file_id])
print(f"Knowledge base ready: {kb_id}")

The file ID parameter is named docIds, not file_ids or fileIds. dataSources is required.

Step 2: Wrap the Retrieval Client

Wrap the Knowledge Studio knowledge search API into a lightweight class for the middleware to call. The knowledge search API works with a published search service (agent); retrieval strategies (rerank, top_k, etc.) are configured in the console, and the API call only requires agent_id and the query.

import aiohttp


class KnowledgeStudioRetriever:
    """Knowledge Studio retrieval client (uses knowledge/search API)."""

    def __init__(self, api_key: str, base_url: str, agent_id: str):
        self.api_key = api_key
        self.base_url = base_url
        self.agent_id = agent_id  # Knowledge search service ID
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    async def search(self, query: str, top_n: int = 5) -> list[dict]:
        """Search knowledge base, return chunks with score and text.

        top_n truncates results locally; retrieval strategy is configured in the agent.
        """
        url = f"{self.base_url}/api/v1/indices/knowledge/search"
        payload = {"agent_id": self.agent_id, "query": query}

        async with aiohttp.ClientSession() as session:
            async with session.post(url, headers=self.headers,
                                    json=payload) as resp:
                data = await resp.json()
                nodes = data["data"]["nodes"]
                return nodes[:top_n]

    def format_context(self, nodes: list[dict]) -> str:
        """Format retrieval results into prompt context text."""
        parts = []
        for n in nodes:
            doc_name = n.get("metadata", {}).get("doc_name", "Unknown")
            parts.append(f"[{doc_name}]\n{n['text']}")
        return "\n---\n".join(parts)

Step 3: Implement KnowledgeStudioRAGMiddleware

This is the core of the recipe — a custom MiddlewareBase subclass implementing both static and agentic modes. Modeled after AgentScope's built-in RAGMiddleware, but with the retrieval backend changed to the Knowledge Studio API.

from typing import Any
from agentscope.middleware import MiddlewareBase
from agentscope.agent import Agent
from agentscope.tool import ToolBase
from agentscope.tool._response import ToolChunk, ToolResultState
from agentscope.permission._decision import PermissionDecision, PermissionBehavior
from agentscope.permission._context import PermissionContext
from agentscope.message import TextBlock


class KnowledgeStudioRAGMiddleware(MiddlewareBase):
    """RAG middleware connecting Knowledge Studio to AgentScope Agent.

    mode="static":  Retrieve before reasoning, inject into system prompt
    mode="agentic": Expose search_knowledge tool for model to call on demand
    """

    def __init__(self, retriever: KnowledgeStudioRetriever,
                 mode: str = "agentic", top_k: int = 5):
        self.retriever = retriever
        self.mode = mode
        self.top_k = top_k

    # ── static mode: on_system_prompt hook ──────────────────────

    async def on_system_prompt(self, agent: Agent,
                               current_prompt: str) -> str:
        """static mode: inject retrieval results into system prompt.

        AgentScope calls this hook (Transformer type, serial pipeline)
        when assembling the system prompt in each ReAct reasoning step.
        """
        if self.mode != "static":
            return current_prompt

        # Extract query from the latest user message in context
        recent_msgs = agent.state.context[-4:]
        user_query = ""
        for msg in reversed(recent_msgs):
            if msg.role == "user":
                user_query = msg.get_text_content()
                break
        if not user_query:
            return current_prompt

        # Retrieve and format
        nodes = await self.retriever.search(user_query, self.top_k)
        if not nodes:
            return current_prompt

        context = self.retriever.format_context(nodes)
        return (f"{current_prompt}\n\n## Knowledge Base Reference\n"
                f"The following relevant content was retrieved. "
                f"Prioritize it when answering:\n\n{context}\n")

    # ── agentic mode: list_tools hook ───────────────────────────

    async def list_tools(self) -> list[ToolBase]:
        """agentic mode: expose search_knowledge tool for model to call.

        AgentScope does not auto-invoke this hook's returned tools —
        collect them manually and pass to Toolkit when assembling the Agent.        """
        if self.mode != "agentic":
            return []

        retriever = self.retriever  # capture for closure

        class SearchKnowledgeTool(ToolBase):
            """Search the knowledge base for relevant information."""
            name: str = "search_knowledge"
            description: str = (
                "Search the knowledge base for relevant information. "
                "Use this when you need to look up facts, docs, or "
                "specific knowledge to answer the question.")
            input_schema: dict = {
                "type": "object",
                "properties": {
                    "query": {"type": "string",
                              "description": "The search query."},
                    "top_k": {"type": "integer", "default": 5,
                               "description": "Number of results to return."},
                },
                "required": ["query"],
            }
            is_concurrency_safe: bool = True
            is_read_only: bool = True

            async def check_permissions(
                self, tool_input: dict[str, Any],
                context: PermissionContext,
            ) -> PermissionDecision:
                return PermissionDecision(
                    behavior=PermissionBehavior.ALLOW,
                    message="auto-allow search")

            async def call(self, query: str, top_k: int = 5):
                """Search the knowledge base, return relevant chunks."""
                nodes = await retriever.search(query, top_k)
                if not nodes:
                    yield ToolChunk(
                        content=[TextBlock(text="No relevant results found.")],
                        state=ToolResultState.SUCCESS,
                        is_last=True,
                    )
                    return
                context = retriever.format_context(nodes)
                yield ToolChunk(
                    content=[TextBlock(text=context)],
                    state=ToolResultState.SUCCESS,
                    is_last=True,
                )

        return [SearchKnowledgeTool()]

AgentScope's ToolBase is an abstract base class. Custom tools need to implement:

Class attributes name, description, input_schema (JSON Schema format)
Class attributes is_concurrency_safe, is_read_only (permission & concurrency control)
check_permissions method (returns PermissionDecision, search tools can simply ALLOW)
call method (async generator, yield ToolChunk to return results)

Step 4: Assemble and Run the Agent

Assemble an AgentScope Agent with DashScopeChatModel + KnowledgeStudioRAGMiddleware, demonstrating both modes end-to-end.

Create Model and Agent

import asyncio
import os
from agentscope.agent import Agent
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential
from agentscope.tool import Toolkit
from agentscope.message import UserMsg

# 1. Create DashScope Chat Model
chat_model = DashScopeChatModel(
    credential=DashScopeCredential(api_key=os.environ["DASHSCOPE_API_KEY"]),
    model="qwen-plus",
    stream=True,
)

# 2. Create retrieval client
retriever = KnowledgeStudioRetriever(
    api_key=os.environ["DASHSCOPE_API_KEY"],
    base_url=BASE_URL,
    agent_id="aid-xxxxxxxx",  # From console knowledge search service
)

Mode 1: Static Injection

async def run_static():
    """static mode: auto-retrieve and inject before each reasoning."""
    mw = KnowledgeStudioRAGMiddleware(retriever, mode="static", top_k=3)
    agent = Agent(
        name="kb-assistant",
        system_prompt=(
            "You are a knowledge base assistant. Answer questions based on "
            "the retrieved knowledge base content. If no relevant information "
            "is found, state that you cannot answer."
        ),
        model=chat_model,
        middlewares=[mw],
    )

    result = await agent.reply(
        UserMsg(name="user", content="How to call the Qianwen API?")
    )
    print(result.get_text_content())

asyncio.run(run_static())

Test output (based on a knowledge base with technical docs):

Steps to call the Qianwen API (based on knowledge base content):

1. Get API Key
   Go to the Alibaba Cloud Bailian platform, enable model invocation service,
   and create an API Key…

2. Install SDK
   pip install dashscope

3. Call API
   Use dashscope.Generation API to make calls…

Execution flow:

AgentScope calls on_system_prompt before each ReAct reasoning step. The middleware auto-retrieves and injects results into the system prompt — the model never needs to know about the retrieval process.

Mode 2: Agentic Retrieval

async def run_agentic():
    """agentic mode: model autonomously decides when and what to retrieve."""
    mw = KnowledgeStudioRAGMiddleware(retriever, mode="agentic", top_k=3)
    tools = await mw.list_tools()  # Manually collect tools provided by middleware

    agent = Agent(
        name="kb-assistant",
        system_prompt=(
            "You are a knowledge base assistant. You can call the "
            "search_knowledge tool to retrieve from the knowledge base. "
            "Answer based on retrieval results."
        ),
        model=chat_model,
        toolkit=Toolkit(tools=tools),
        middlewares=[mw],
    )

    result = await agent.reply(
        UserMsg(name="user", content="How to call the Qianwen API?")
    )
    print(result.get_text_content())

asyncio.run(run_agentic())

Test output:

There are several ways to call the Qianwen API:

### 1. OpenAI Compatible Interface (Recommended)
For developers familiar with the OpenAI SDK…

### 2. DashScope SDK
Using Alibaba Cloud's native SDK…

### 3. Direct HTTP Calls
Using curl or any HTTP client…

Execution flow:

The model autonomously decides whether to retrieve during the ReAct loop — simple questions may be answered directly; questions requiring documentation trigger a tool call with a self-constructed query.

Combining Both Modes

AgentScope allows mounting two instances with different modes — both auto-inject and on-demand tool:

static_mw = KnowledgeStudioRAGMiddleware(retriever, mode="static", top_k=3)
agentic_mw = KnowledgeStudioRAGMiddleware(retriever, mode="agentic", top_k=5)
tools = await agentic_mw.list_tools()

agent = Agent(
    name="kb-assistant",
    system_prompt="You are a knowledge base assistant.",
    model=chat_model,
    toolkit=Toolkit(tools=tools),
    middlewares=[static_mw, agentic_mw],
)

Mode Comparison and Selection

Dimension	Static Injection	Agentic Retrieval	Combined
Retrieval decision	Fixed (always before reasoning)	Model decides (on demand)	Both
AgentScope hook	`on_system_prompt`	`list_tools`	Both
Context length	top_k chunks injected each time	Tool results on demand	Combined
Multi-round retrieval	No (one-shot)	Yes (model can call tool multiple times)	Yes
Best for	FAQ, clear-intent queries	Complex QA, needs judgment	High-reliability
vs. built-in RAGMiddleware	Same design (static mode)	Same design (agentic mode)	Same design

This custom middleware is designed to be consistent with AgentScope's built-in RAGMiddleware — same two modes, same parameter naming (mode, top_k). The difference is the retrieval backend: the built-in RAGMiddleware works with a local KnowledgeBase (embedding model and vector store), while this middleware calls the Knowledge Studio knowledge search API directly.

Extension: Integrate with AgentScope Persistent State

AgentScope's AgentState can be serialized to JSON and stored in Redis, enabling cross-session state recovery. The knowledge base ID persists as middleware configuration alongside the Agent:

from agentscope.state import AgentState
from agentscope.app.storage import RedisStorage

USER_ID = "user_123"
AGENT_ID = "agent_456"
SESSION_ID = "session_789"

async def run_with_persistence():
    async with RedisStorage(host="localhost", port=6379) as storage:
        record = await storage.get_session(USER_ID, AGENT_ID, SESSION_ID)
        state = record.state if record else AgentState()

        agent = Agent(
            name="kb-assistant",
            system_prompt="You are a knowledge base assistant.",
            model=chat_model,
            toolkit=Toolkit(tools=await mw.list_tools()),
            middlewares=[mw],
            state=state,
        )

        result = await agent.reply(
            UserMsg(name="user", content="Is the feature mentioned last time still supported?")
        )

        await storage.update_session_state(
            USER_ID, AGENT_ID, SESSION_ID, agent.state)

Knowledge Studio retrieval is stateless — each search call is independent, the server does not preserve conversation context. Conversation history is managed by AgentScope's AgentState, decoupled from knowledge base retrieval. For multi-turn conversation with tool history, see the Multi-Turn Chat cookbook.

FAQ

Why not use AgentScope's built-in RAGMiddleware?

The built-in RAGMiddleware works with local KnowledgeBase objects, requiring a local embedding model and vector store. Knowledge Studio is a managed service where embedding, vector storage, and retrieval all run server-side. This middleware calls the knowledge search API directly, eliminating the need to deploy and maintain a local vector store.

Is static mode too slow (retrieves every reasoning step)?

The Knowledge Studio knowledge search API typically responds in 200-500ms. If latency is a concern, switch to agentic mode for on-demand retrieval, or reduce top_k. With top_k=3, the static mode Agent responds in 3-5 seconds (including model generation time).

What if the model doesn't call the tool in agentic mode?

Ensure the system prompt explicitly mentions "you can call the search_knowledge tool to retrieve from the knowledge base." In testing, qwen-plus autonomously calls the tool when the question involves knowledge base content. If the model still doesn't call it, try a more capable model (e.g., qwen-max).

Can I use both Knowledge Studio chat API and AgentScope?

Yes, but they are alternatives, not complementary. Knowledge Studio chat API is a managed agent service with built-in retrieval; AgentScope is a framework for building custom agents. Use chat API when you don't need a custom agent; use AgentScope when you need custom tool chains, middleware, or human-in-the-loop.

​Principle: Knowledge Base as Agent Memory

​Prerequisites

​Install Dependencies

​Required Parameters

​Step 1: Build Knowledge Base via API

​Step 2: Wrap the Retrieval Client

​Step 3: Implement KnowledgeStudioRAGMiddleware

​Step 4: Assemble and Run the Agent

​Create Model and Agent

​Mode 1: Static Injection

​Mode 2: Agentic Retrieval

​Combining Both Modes

​Mode Comparison and Selection

​Extension: Integrate with AgentScope Persistent State

​FAQ

​Why not use AgentScope's built-in RAGMiddleware?

​Is static mode too slow (retrieves every reasoning step)?

​What if the model doesn't call the tool in agentic mode?

​Can I use both Knowledge Studio chat API and AgentScope?

Principle: Knowledge Base as Agent Memory

Prerequisites

Install Dependencies

Required Parameters

Step 1: Build Knowledge Base via API

Step 2: Wrap the Retrieval Client

Step 3: Implement KnowledgeStudioRAGMiddleware

Step 4: Assemble and Run the Agent

Create Model and Agent

Mode 1: Static Injection

Mode 2: Agentic Retrieval

Combining Both Modes

Mode Comparison and Selection

Extension: Integrate with AgentScope Persistent State

FAQ

Why not use AgentScope's built-in RAGMiddleware?

Is static mode too slow (retrieves every reasoning step)?

What if the model doesn't call the tool in agentic mode?

Can I use both Knowledge Studio chat API and AgentScope?