示例代码

OpenAPI操作

参考官方文接口文档

OpenAPI

检索结果：

参考代码python：

import os

from ksyun.common import credential
from ksyun.common.profile.client_profile import ClientProfile
from ksyun.common.profile.http_profile import HttpProfile
from ksyun.common.exception.ksyun_sdk_exception import KsyunSDKException
from ksyun.client.aicp.v20251114 import client as aicp_client, models as aicp_models

try:
    cred = credential.Credential(
        os.environ.get("KSYUN_SECRET_ID", "AK here"),##此处填AK
        os.environ.get("KSYUN_SECRET_KEY", "SK here")##此处填SK
    )

    httpProfile = HttpProfile()
    httpProfile.endpoint = "aicp.inner.api.ksyun.com"#内部账号请求该地址|公网请求aicp.api.ksyun.com
    httpProfile.reqMethod = "POST"
    httpProfile.reqTimeout = 60
    httpProfile.scheme = "http"
    clientProfile = ClientProfile()
    clientProfile.httpProfile = httpProfile

    aicpClient = aicp_client.AicpClient(cred, "cn-north-vip1", profile=clientProfile)
    print(aicpClient.call_json("RetrieveKnowledge", {
            "DatasetId": "e50168a9-354a-4f3f-9a26-d7899d058201",##此处填你知识库相应的实例ID
            "Query": "金山云AgentEngine是什么呢",#问题的query
            "RetrievalModel": {
                "SearchMethod": "intelligence_search",
                "RerankingEnable": True,
                "RerankingMode": {
                    "RerankingProviderName": "1",
                    "RerankingModelName": "Qzhou/BGE-Reranker-v2-m3"
                },
                "TopK": 1,
                "ScoreThresholdEnabled": True,
                "ScoreThreshold": 0.5,
                "Retriever": {
                    "Vector": {
                        "TopK": 1,
                        "ScoreThreshold": 0.5,
                        "ScoreThresholdEnabled": True
                    },
                    "Inverted": {
                        "TopK": 1,
                        "ScoreThreshold": 0.5,
                        "ScoreThresholdEnabled": True
                    }
                }
            }
        }))
except KsyunSDKException as err:
    print(err)

基于LangChain框架搭建使用知识库

演示结果：

参考代码：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
LangChain 知识检索 Agent 模板
基于金山云 AICP RetrieveKnowledge API
使用 langgraph 构建
"""
import os
import sys
import io
import json

# Windows UTF-8 支持
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

from ksyun.common import credential
from ksyun.common.profile.client_profile import ClientProfile
from ksyun.common.profile.http_profile import HttpProfile
from ksyun.client.aicp.v20251114 import client as aicp_client


# ==================== 配置 ====================

# 金山云 AICP
KSYUN_SECRET_ID = os.environ.get("KSYUN_SECRET_ID", "AK Here")##此处填写AK
KSYUN_SECRET_KEY = os.environ.get("KSYUN_SECRET_KEY", "SK Here")##此处填写SK
KSYUN_ENDPOINT = os.environ.get("KSYUN_ENDPOINT", "aicp.inner.api.ksyun.com")##内部账号请求该地址|公网请求aicp.api.ksyun.com
DATASET_ID = os.environ.get("DATASET_ID", "XXX")##知识库ID

# LLM
LLM_API_KEY = os.environ.get("LLM_API_KEY", "XXX")##大模型的Key
LLM_MODEL = os.environ.get("LLM_MODEL", "XXX")##模型种类
LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "XXX")##模型请求的URL


# ==================== 工具定义 ====================

@tool
def retrieve_knowledge(query: str) -> str:
    """从知识库检索相关信息。输入用户问题，返回知识内容。"""
    try:
        cred = credential.Credential(KSYUN_SECRET_ID, KSYUN_SECRET_KEY)

        httpProfile = HttpProfile()
        httpProfile.endpoint = KSYUN_ENDPOINT
        httpProfile.reqMethod = "POST"
        httpProfile.reqTimeout = 60
        httpProfile.scheme = "http"

        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile

        aicpClient = aicp_client.AicpClient(cred, "cn-north-vip1", profile=clientProfile)

        result = aicpClient.call_json("RetrieveKnowledge", {
            "DatasetId": DATASET_ID,
            "Query": "金山云AgentEngine是什么呢",
            "RetrievalModel": {
                "SearchMethod": "intelligence_search",
                "RerankingEnable": True,
                "TopK": 3,
                "ScoreThreshold": 0.5,
                "ScoreThresholdEnabled": True
            }
        })

        if isinstance(result, str):
            result = json.loads(result)

        contents = []
        for record in result.get("Records", []):
            content = record.get("Segment", {}).get("Content")
            if content:
                score = record.get("Score", 0)
                contents.append(f"[相关度:{score:.2f}]\n{content}")

        return "\n\n---\n\n".join(contents) if contents else "未找到相关内容"

    except Exception as e:
        return f"检索失败: {e}"


# ==================== Agent 创建 ====================

def create_agent():
    """创建 Agent"""
    llm = ChatOpenAI(
        model=LLM_MODEL,
        api_key=LLM_API_KEY,
        base_url=LLM_BASE_URL,
        temperature=0.7
    )

    return create_react_agent(
        model=llm,
        tools=[retrieve_knowledge],
        prompt="你是智能助手，使用 retrieve_knowledge 工具检索知识库回答问题。需要根据检索出来的结果，进行回答，要求不胡编乱造不创造虚假信息"
    )


# ==================== 主程序 ====================

def main():
    agent = create_agent()

    print("=" * 50)
    print("   知识检索 Agent (LangGraph)")
    print("=" * 50)
    print("输入 'quit' 退出\n")

    while True:
        try:
            query = input("用户: ").strip()
            if not query or query.lower() in ['quit', 'exit', 'q']:
                break

            result = agent.invoke({"messages": [("user", query)]})
            # 获取最后一条消息
            last_msg = result["messages"][-1]
            print(f"\n助手: {last_msg.content}\n")

        except KeyboardInterrupt:
            break

    print("\n再见!")


if __name__ == '__main__':
    main()

基于AgentEngine框架和KsADK使用知识库

配置环境变量

# 配置环境变量
## 创建好 Knowledge 后，请选择以下任意一种方式完成环境变量
# 方式一：通过智能体运行时配置页关联该知识库
# 方式二：通过以下环境变量配置
# 如果在 Runtime 配置运行记忆库 + VeADK，不需要配置 AK/SK； 其他使用方式需要配置 AK/SK
OPENAI_API_KEY=xxx
OPENAI_API_BASE=http://kspmas.ksyun.com/v1
KSADK_KB_DATASET_ID=你的知识库ID    # 只要这个存在，工具就自动注入
KSYUN_ACCESS_KEY=你的AK
KSYUN_SECRET_KEY=你的SK

# LLM 配置
OPENAI_API_BASE=https://kspmas.ksyun.com/v1/  #模型url，默认使用星流平台的模型
OPENAI_API_KEY=你的key
MODEL_NAME=模型名称

安装Python依赖

# Python依赖安装
pip install ksadk

导入知识库&模型配置

"""
知识检索 Agent 示例 (使用 Google ADK + ksadk)
基于 ksadk - 金山云 Agent 开发工具包
"""
import asyncio
import os
import sys
import io

# Windows UTF-8 支持
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

from dotenv import load_dotenv

load_dotenv()  # 从 .env 加载配置

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# 显式导入知识库工具
from ksadk.knowledge_base.adk_tool import search_knowledge_base


# ==================== 配置 ====================

# 从 .env 加载模型配置
MODEL_NAME = os.getenv("MODEL_NAME", "deepseek-v3-0324")
OPENAI_API_BASE = os.getenv("OPENAI_API_BASE", "https://kspmas.ksyun.com/v1/")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")

# 模型配置
model = LiteLlm(
    model=f"openai/{MODEL_NAME}",
    api_base=OPENAI_API_BASE,
    api_key=OPENAI_API_KEY,
)


# ==================== Agent 创建 ====================

# Agent — 显式添加知识库工具
agent = Agent(
    name="kb_assistant",
    model=model,
    description="知识库问答助手",
    instruction="""你是一个知识库问答助手。
当用户提问时，先调用 search_knowledge_base 工具检索知识库，
然后基于检索结果回答。回答中标注信息来源文档名。用中文回答。""",
    tools=[search_knowledge_base],
)

# Runner 初始化
session_service = InMemorySessionService()
runner = Runner(agent=agent, session_service=session_service, app_name="kb_demo")


async def chat(question: str):
    """发送问题并获取回答"""
    session = await session_service.create_session(app_name="kb_demo", user_id="user1")
    msg = types.Content(role="user", parts=[types.Part(text=question)])

    final = ""
    async for event in runner.run_async(
        user_id="user1", session_id=session.id, new_message=msg
    ):
        if (
            event.content
            and event.content.parts
            and hasattr(event.content.parts[0], "text")
            and event.content.parts[0].text
        ):
            if not getattr(event.content.parts[0], "thought", False):
                final = event.content.parts[0].text.strip()

    print(f"Q: {question}")
    print(f"A: {final}")


if __name__ == "__main__":
    asyncio.run(chat("AgentEngine是什么东西呢"))