知识库集成指南

绑定阿里云百炼知识库

如果您使用阿里云百炼作为 RAG 服务，请配置以下参数：

• 知识库名称：输入自定义的知识库显示名称。

• 描述：简述该知识库的用途或涵盖的数据范围。

• 知识库类型：选择百炼。

• 工作空间：首次使用时，点击默认业务空间的立即授权按钮，开启授权并确认业务空间。

• 知识库：从下拉列表中选择已在阿里云百炼知识库创建好的知识库（支持多选），并根据业务需要修改检索配置。

点击创建知识库，即可将您的阿里云百炼知识库绑定至AgentRun平台。

绑定 RAGFlow 知识库

RAGFlow 是一款开源的深度文档理解引擎，绑定时请填写以下信息：

• 知识库名称：输入自定义的知识库显示名称。

• 描述：简述该知识库的用途或涵盖的数据范围。

• 知识库类型：选择RAGFlow。

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

  • BaseURL：填写 RAGFlow 服务的 API 端点地址。

  • DataSet IDs：填写需要关联的数据集 ID（支持多个 ID）。

  • 凭证配置：选择已保存的RAGFlow的API-KEY凭证。

  说明

  关于如何获取上述 RAGFlow 配置参数，请参考本文如何获取 RAGFlow 配置信息。

• 检索配置：根据业务需要修改相似度阈值和向量相似度权重。

点击创建知识库，即可将您自建的RAGFlow知识库绑定至AgentRun平台。

获取 BaseURL

• 点击右上角用户头像，进入个人设置/账户管理页面。

• 找到API部分，查看基础请求地址API Server即为BaseURL。

  

获取 API-KEY

• 在个人API设置页面中，找到RAGFlow API > API KEY。

• 复制已有的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/quick-start 配置好相应认证信息和环境变量

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()