Agent Run：知識庫整合指南

綁定阿里雲百鍊知識庫

如果您使用阿里雲百鍊作為 RAG 服務，請配置以下參數：

• 知識庫名稱：輸入自訂的知識庫顯示名稱。

• 描述：簡述該知識庫的用途或涵蓋的資料範圍。

• 知識庫類型：選擇百鍊。

• 工作空間：首次使用時，點擊預設業務空間的立即授權按鈕，開啟授權並確認業務空間。

• 知識庫：從下拉式清單中選擇已在阿里雲百鍊知識庫建立好的知識庫（支援多選），並根據業務需要修改檢索配置。

點擊建立知識庫，即可將您的阿里雲百鍊知識庫綁定至AgentRun平台。

綁定 RAGFlow 知識庫

RAGFlow 是一款開源的深度文檔理解引擎，綁定時請填寫以下資訊：

• 知識庫名稱：輸入自訂的知識庫顯示名稱。

• 描述：簡述該知識庫的用途或涵蓋的資料範圍。

• 知識庫類型：選擇RAGFlow。

• RAGFlow配置（可參考控制台指引，快速通過SAE部署RAGFlow）：

  • BaseURL：填寫 RAGFlow 服務的 API 端點地址。

  • DataSet IDs：填寫需要關聯的資料集 ID（支援多個 ID）。

  • 憑證配置：選擇已儲存的RAGFlow的API-KEY憑證。

  說明

  關於如何擷取上述 RAGFlow 配置參數，請參考本文如何擷取 RAGFlow 配置資訊。

• 檢索配置：根據業務需要修改相似性閾值和向量相似性權重。

點擊建立知識庫，即可將您自建的RAGFlow知識庫綁定至AgentRun平台。

建立 Tablestore 知識庫

如果您沒有現成的向量資料庫，可以選擇 Tablestore 知識庫，由 AgentRun 負責執行個體的建立與索引配置。在知識庫列表頁單擊建立知識庫，接入方式選擇建立知識庫，然後按以下兩種配置方式之一建立：

• 自動設定：固定使用 Tablestore，無需選擇知識庫類型和執行個體。提交前 AgentRun 會檢查 Tablestore 服務的開通狀態，並檢查 AliyunOTSAccessingOSSRole 角色；角色不存在時會中止建立並引導您完成授權。授權完成後，AgentRun 自動建立一個名稱形如 knowledge-xxxx、叢集類型為 SSD 的 Tablestore 執行個體，並為其打上 source=agentrun 和 scene=knowledgebase 標籤。

• 自訂配置：知識庫類型選擇 Tablestore 知識庫，並從 Tablestore 執行個體下拉式清單中選擇當前地區已有的執行個體。

配置以下參數：

在檢索配置中設定知識庫的預設檢索行為，建立後可在知識庫詳情頁調整：

提交後，AgentRun 先在 Tablestore 中建立知識庫，再在 AgentRun 中產生對應的知識庫記錄。建立完成後，可在知識庫詳情頁匯入文件並驗證召回效果，參見下文「自建知識庫的文件管理與召回測試」。

關於 Embedding 模型的接入與配置，請參見向量模型-添加與配置 Embedding 模型。

建立 AnalyticDB PostgreSQL 知識庫

如果您已有 AnalyticDB PostgreSQL 執行個體，並希望知識庫資料儲存在自有執行個體中，可以選擇 AnalyticDB PostgreSQL 知識庫。建立前請確認：已有可用的 AnalyticDB PostgreSQL 執行個體、可訪問該執行個體的帳號與密碼、以及執行個體中已建立的命名空間及其密碼（系統命名空間不會出現在候選列表中）。

在建立知識庫面板中依次選擇：建立方式為建立知識庫、配置方式為自訂配置、知識庫類型為 AnalyticDB PostgreSQL 知識庫，然後配置以下參數：

使用者名稱、密碼、中繼資料配置、Embedding Model 和 Index Algorithm 均在選擇執行個體後才展示。

在檢索配置中設定知識庫的預設檢索行為，建立後可在知識庫詳情頁調整：

提交後，AgentRun 先在 AnalyticDB PostgreSQL 執行個體的指定命名空間中建立文檔集合，再在 AgentRun 中產生對應的知識庫記錄。建立完成後，可在知識庫詳情頁匯入文件並驗證召回效果，參見下文「自建知識庫的文件管理與召回測試」。

擷取 BaseURL

• 點擊右上方帳戶圖片，進入個人設定/賬戶管理頁面。

• 找到API部分，查看基礎請求地址API Server即為BaseURL。

擷取 API-KEY

• 在個人API設定頁面中，找到。

• 複製已有的API KEY，如果沒有建立過，點擊Create new key按鈕進行建立。

• 參考憑證管理將擷取到的API KEY作為訪問第三方的憑證綁定在AgentRun平台，注意建立憑證時服務類型選擇其他服務。

擷取 Dataset ID

• 點擊頂部功能表列的Knowledge Base。

• 在列表中找到您想要關聯的資料集，點擊進入。

• 檢查瀏覽器地址欄的 URL，或在頁面的配置資訊列中找到該資料集的 UUID，即為 Dataset ID。

知識庫查詢程式碼範例

使用 agentrun.knowledgebase 模組，可以直接檢索知識庫內容。

from agentrun.knowledgebase import KnowledgeBase

## 擷取單知識庫，進行查詢
knowledgebase = KnowledgeBase.get_by_name("ragflow-test")
single_kb_retrieve_result = knowledgebase.retrieve("<your-query>")
print(single_kb_retrieve_result)

## 擷取多知識庫，進行查詢，支援跨供應商知識庫類型檢索
multi_kb_retrieve_result = KnowledgeBase.multi_retrieve(
    query="<your-query>",
    knowledge_base_names=["ragflow-test", "<your-knowledge-base-name-2>"],
)
print(multi_kb_retrieve_result)

在 Agent 架構中整合知識庫

您可以將知識庫作為“工具”，注入到 LangChain 或其他 Agent 架構中，使 AI 具備自動查閱文檔的能力。

"""AgentRun 知識庫智能體整合程式碼範例

使用前，請參考https://docs.agent.run/docs/tutorial/quickstart/manual 配置好相應認證資訊和環境變數

curl http://127.0.0.1:9000/openai/v1/chat/completions -X POST \
    -H "Content-Type: application/json" \
    -d '{"messages": [{"role": "user", "content": "什麼是Serverless?"}], "stream": true}'
"""

import json
import os
from typing import Any

from langchain.agents import create_agent
import pydash

from agentrun import Config
from agentrun.integration.langchain import model
from agentrun.integration.langchain import knowledgebase_toolset
from agentrun.integration.langgraph.agent_converter import AgentRunConverter
from agentrun.knowledgebase import KnowledgeBase
from agentrun.server import AgentRequest, AgentRunServer
from agentrun.server.model import ServerConfig
from agentrun.utils.log import logger

# 請替換為您已經建立的 模型 名稱
AGENTRUN_MODEL_SERVICE = os.getenv("AGENTRUN_MODEL_SERVICE", "<your-model-service>")
AGENTRUN_MODEL_NAME = os.getenv("AGENTRUN_MODEL_NAME", "<your-model-name>")
KNOWLEDGE_BASES = os.getenv("AGENTRUN_KNOWLEDGE_BASES", "ragflow-test").split(",")

if AGENTRUN_MODEL_NAME.startswith("<") or not AGENTRUN_MODEL_NAME:
    raise ValueError("請將 MODEL_NAME 替換為您已經建立的模型名稱")

## 載入知識庫工具，知識庫可以以工具的方式供Agent進行調用
knowledgebase_tools = []
if KNOWLEDGE_BASES and not KNOWLEDGE_BASES[0].startswith("<"):
    knowledgebase_tools = knowledgebase_toolset(
        knowledge_base_names=KNOWLEDGE_BASES,
    )
else:
    logger.warning("KNOWLEDGE_BASES 未設定或未替換，跳過載入知識庫工具。")

agent = create_agent(
    model=model(AGENTRUN_MODEL_SERVICE, model=AGENTRUN_MODEL_NAME, config=Config(timeout=180)),
    tools=[
        *knowledgebase_tools,   ## 通過工具整合知識庫查詢能力
    ],
    system_prompt="你是一個 AgentRun 的 AI 專家，可以通過查詢知識庫文檔來回答使用者的問題。",
)


async def invoke_agent(request: AgentRequest):
    messages = [
        {"role": msg.role, "content": msg.content}
        for msg in request.messages
    ]

    # 如果配置了知識庫，查詢知識庫並將結果添加到上下文
    if KNOWLEDGE_BASES and not KNOWLEDGE_BASES[0].startswith("<"):
        # 擷取使用者最新的訊息內容作為查詢
        user_query = None
        for msg in reversed(request.messages):
            if msg.role == "user":
                user_query = msg.content
                break

        if user_query:
            try:
                retrieve_result = await KnowledgeBase.multi_retrieve_async(
                    query=user_query,
                    knowledge_base_names=KNOWLEDGE_BASES,
                )
                # 直接將檢索結果添加到上下文
                if retrieve_result:
                    messages.append({
                        "role": "assistant",
                        "content": json.dumps(retrieve_result, ensure_ascii=False),
                    })
            except Exception as e:
                logger.warning(f"知識庫檢索失敗: {e}")

    input: Any = {"messages": messages}

    converter = AgentRunConverter()
    if request.stream:

        async def async_generator():
            async for event in agent.astream(input, stream_mode="updates"):
                for item in converter.convert(event):
                    yield item

        return async_generator()
    else:
        result = await agent.ainvoke(input)
        return pydash.get(result, "messages[-1].content", "")


AgentRunServer(
    invoke_agent=invoke_agent,
    config=ServerConfig(
        cors_origins=[
            "*"
        ]
    ),
).start()