全部文档
当前文档
最近更新时间：2026-03-06 10:47:19
# 环境变量配置
## 创建好 Knowledge 后，请选择以下任意一种方式完成环境变量
# 方式一：通过智能体运行时配置页关联该知识库
# 方式二：通过以下环境变量配置
# 如果在 Runtime 配置运行记忆库 + VeADK，不需要配置 AK/SK； 其他使用方式需要配置 AK/SK
export KSADK_KB_DATASET_ID=你的知识库ID
export KSYUN_ACCESS_KEY=你的AK
export KSYUN_SECRET_KEY=你的SK
# Python依赖安装
pip install ksadk
# Python代码集成
# 参考如下 Knowledge 示例集成代码，修改 agent 代码
import asyncio
import os
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
# 模型配置
model = LiteLlm(
model=f"openai/{os.getenv('MODEL_NAME', 'deepseek-v3.2')}",
api_base=os.getenv("OPENAI_API_BASE"),
api_key=os.getenv("OPENAI_API_KEY"),
)
# 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("如何配置知识库？"))纯净模式