实战教程：从零搭建自主查资料的AI智能体

从 0 到 1 搭一个能『自己查资料』的 AI 智能体

一篇能照着做、做完就能跑起来的实战教程。所有数据都在你电脑上。

一、为什么是 AgentScope？

智能体开发框架现在很多：LangChain、LlamaIndex、AutoGen、CrewAI……如果你需要：

• 开箱即用的 ReAct 范式（思考-行动-观察循环）

• 官方内置知识库抽象（KnowledgeBase+QdrantStore）

• 不写 prompt 模板也能让模型自动调用工具

AgentScope 2.0.x 是一个很好的选择。它是蚂蚁集团开源的多智能体框架，Python API 设计现代（async-first、ContentBlock 列表、统一的 Credential 管理），文档和示例中文友好。

这次要做的事情用一句话概括：

给 AI 一本技术字典，让它回答用户问题前自己翻字典，引用出处再作答。

下面从环境到 UI 一层层拆开，全程可复现。

二、整体架构

项目分四个模块，对应 4 个核心文件：

四个文件不到 600 行代码，跑通完整 RAG 链路。

数据流

用户提问
│
▼
[前端 /api/chat SSE]
│
▼
agent.reply_stream()
│ ┌──────────────────────────────────────┐
├─►│ Thought（思考） │──► SSE: thinking
│ ├──────────────────────────────────────┤
├─►│ Action：调 search_knowledge_base 工具 │──► SSE: tool
│ │ ┌──────────────────────────┐ │
│ │ │ knowledge_base.search() │ │
│ │ │ ↓ Qdrant 向量检索 │ │
│ │ │ ↓ DashScope 实时嵌入 │ │
│ │ └──────────────────────────┘ │
│ ├──────────────────────────────────────┤
├─►│ Observation：检索结果 │──► SSE: sources
│ ├──────────────────────────────────────┤
└─►│ Final Answer：基于结果作答 │──► SSE: token...
│ └──────────────────────────────────────┘
▼
前端流式渲染（token 实时显示、sources 折叠到回答下方）

三、阶段 0：环境准备

# 建隔离 venv
python -m venv .venv
.venv\Scripts\activate

# 安装依赖
pip install agentscope==2.0.4 qdrant-client python-dotenv

确认版本：

python -c "import agentscope; print(agentscope.__version__)"
# → 2.0.4

.env配置：

# 对话模型（DeepSeek）
DEEPSEEK_API_KEY=sk-...
DEEPSEEK_BASE_URL=https://api.deepseek.com
CHAT_MODEL_NAME=deepseek-v4-flash

# 嵌入模型（通义 DashScope）
EMBEDDING_PROVIDER=dashscope
DASHSCOPE_API_KEY=sk-...
DASHSCOPE_EMBEDDING_MODEL=text-embedding-v3
DASHSCOPE_EMBEDDING_DIMENSIONS=1024

# 向量库（本地嵌入式，零运维）
QDRANT_PATH=./qdrant_db

关于模型选型：DeepSeek 目前只提供对话接口，没有 embedding 端点，AgentScope 2.0.4 也不内置DeepSeekEmbeddingModel。所以嵌入层选择 DashScope 的text-embedding-v3（1024 维），对中文支持好、有免费额度。如果你偏好本地部署，也可以换成 Ollama 的嵌入模型，AgentScope 有OllamaEmbeddingModel。

Qdrant 文件锁：本地 path 模式下，同一个./qdrant_db目录不能同时被两个进程打开。开发时注意ingest.py和后续脚本不要并行运行。

四、阶段 1：知识库构建

4.1 语料

corpus/下放 5 篇中文技术文章（.md格式），每篇覆盖一个智能体子主题：

corpus/
├── 01-react-推理行动范式.md
├── 02-rag-检索增强生成.md
├── 03-工具调用与function-calling.md
├── 04-多智能体协作.md
└── 05-智能体记忆机制.md

4.2 ingestion 管线

AgentScope 2.0.4 的KnowledgeBase将解析器、分块器、向量库、嵌入模型都拆成可插拔组件。完整的ingest.py：

from agentscope.rag import TextParser, ApproxTokenChunker
from config import build_knowledge_base

asyncdefingest(reset: bool = False) -> None:
kb = build_knowledge_base(
name="agent_dev_kb",
description="智能体开发技术知识库",
collection="agent_dev_articles",
)
parser = TextParser()
chunker = ApproxTokenChunker(chunk_size=512, overlap=50)

await kb.ensure_collection()

if reset:
for doc inawait kb.list_documents():
await kb.delete_document(doc.document_id)

for fp insorted(CORPUS_DIR.glob("*.md")):
content = fp.read_text(encoding="utf-8")
sections = await parser.parse(content, fp.name) # 1. 解析
chunks = await chunker.chunk(sections) # 2. 分块
await kb.insert_document( # 3. 嵌入 + 入库
chunks,
document_id=fp.stem,
document_metadata={"filename": fp.name},
)

三步 API：parse→chunk→insert_document，全部 async。

注意：检索结果hit.chunk.content的类型是TextBlock 对象（或 TextBlock 列表），不是str。取文本时需做类型判断：

def block_text(c):
if c isNone: return""
ifisinstance(c, str): return c
ifisinstance(c, list):
return"\n".join(
b.text ifhasattr(b, "text") elsestr(b) for b in c
)
returngetattr(c, "text", str(c))

执行导入：

python ingest.py --reset

预期输出：

[ok] 01-react-推理行动范式.md -> 4 chunks
[ok] 02-rag-检索增强生成.md -> 3 chunks
...
导入完成：5 个文档，共 15 个 chunk

五、阶段 2：检索增强 Agent

核心思路：把知识库检索封装成一个FunctionTool，让 Agent 在 ReAct 循环中自主决定何时检索、检索什么。

5.1 FunctionTool 封装

from agentscope.tool import FunctionTool

asyncdef_search_knowledge_base(query: str, top_k: int = 5) -> str:
"""在智能体开发知识库中检索与问题相关的文档片段。"""
kb = build_knowledge_base(KB_NAME, KB_DESC, COLLECTION)
results = await kb.search(query, top_k=top_k)

docs = []
for h in results:
text = block_text(getattr(h.chunk, "content", None))
src = (h.chunk.metadata or {}).get("filename", "未知文档")
score = float(getattr(h, "score", 0.0) or0.0)
docs.append({"source": src, "score": score, "text": text})

global _last_sources
_last_sources.extend(docs)

ifnot docs:
return"未在知识库中找到相关内容。"

return"\n\n".join(
f"[{i}] 来源《{d['source']}》（相关度 {d['score']:.2f}）\n{d['text']}"
for i, d inenumerate(docs, 1)
)

retrieval_tool = FunctionTool(
func=_search_knowledge_base,
name="search_knowledge_base",
description=(
"在「智能体开发技术知识库」中检索相关文档片段。"
"当用户询问 Agent / 智能体、ReAct、RAG、工具调用、"
"Function Calling、多智能体协作、记忆机制等开发相关"
"问题时使用本工具获取权威资料。"
),
)

description写得越具体、给的使用场景越多，模型主动调用工具的概率越高。

5.2 构造 ReAct Agent

from agentscope.agent import Agent, ReActConfig
from agentscope.tool import Toolkit
from agentscope.state import AgentState
from agentscope.permission import PermissionContext, PermissionMode

def_build_agent() -> Agent:
model = get_chat_model()
toolkit = Toolkit(tools=[retrieval_tool])
react_config = ReActConfig(max_iters=5)

state = AgentState(
permission_context=PermissionContext(mode=PermissionMode.BYPASS),
)

return Agent(
name="agent_dev_assistant",
system_prompt=_SYSTEM_PROMPT,
model=model,
toolkit=toolkit,
react_config=react_config,
state=state,
)

权限要点：ReAct 的默认权限模式是REQUIRE_USER_CONFIRM，这会导致每次工具调用前等待人工确认，循环不会自动推进。对只读的检索工具，用PermissionMode.BYPASS让 Agent 自主执行。注意：不要把这个模式用于有副作用（写操作、网络请求修改状态）的工具。

5.3 流式事件消费

AgentScope 2.0.4 的流式接口：

async for evt in agent.reply_stream(user_msg):
t = str(getattr(evt, "type", "")).lower()
if t == "text_block_delta": yield ("token", evt.delta)
elif t == "thinking_block_delta": yield ("thinking", evt.delta)
elif t == "tool_call_start": yield ("tool", evt.tool_call_name)

事件type字段是大写字符串（TEXT_BLOCK_DELTA、THINKING_BLOCK_DELTA、TOOL_CALL_START），比对前先.lower()。

stream_reply生成器在结束时 yield("sources", deduped)+("done", None)，供 UI 区分"正在生成"和"生成完毕"。

六、阶段 3：UI 层

Agent 写好后用命令行交互当然可以跑。但要做展示或分享，一个好看的可视化界面会让体验完全不同。这里没有使用 Gradio——它的默认样式定制空间有限，视觉效果比较单调。取而代之的是Python 标准库 + 自建前端的组合：

6.1 后端：server.py

整个server.py不到 200 行。关键设计：用后台 daemon 线程运行一个持久事件循环，所有对话请求通过run_coroutine_threadsafe提交。

import asyncio, json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# 持久事件循环：AgentScope 的流式生成器依赖长期存活的 loop
_LOOP = asyncio.new_event_loop()
def_run_loop():
asyncio.set_event_loop(_LOOP)
_LOOP.run_forever()
threading.Thread(target=_run_loop, daemon=True).start()

classHandler(BaseHTTPRequestHandler):
defdo_POST(self):
ifself.path == "/api/chat":
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.end_headers()
fut = asyncio.run_coroutine_threadsafe(pump(), _LOOP)
fut.result(timeout=180)

设计决策说明：为什么用持久 loop 而不是每请求新建？AgentScope 内部（Agent._reasoning、httpxAsyncStream）维护的异步任务生命周期跨越整个 ReAct 循环。每次请求新建 + 销毁 loop 会导致流式协程在中途挂起。用一个 daemon 线程跑run_forever()的持久 loop，和 Gradio 内部 uvicorn 的行为一致。

API 接口：

• GET /api/stats— 知识库统计（文档数、就绪状态）

• POST /api/reset— 清空多轮对话历史

• POST /api/chat— SSE 流，事件类型：token/thinking/tool/sources/done/error

其他几个容易忽略的细节：

• 服务器绑0.0.0.0而非127.0.0.1。Windows 上部分浏览器把localhost解析为 IPv6::1，只绑 IPv4 地址会导致连接被拒、前端报 "failed to fetch"

• /api/stats内部的异步调用也要走run_coroutine_threadsafe，不能直接在已运行run_forever()的 loop 上调run_until_complete

• 确保用项目 venv 的 Python 启动（而非系统或托管 Python），否则会缺少agentscope/qdrant_client等包

6.2 前端：index.html

设计目标：看得像专业聊天产品，同时保持代码简洁（单文件、零依赖）。

侧边栏（288px 固定宽度）：

• 渐变品牌 logo + 应用描述

• 知识库状态卡（文档数 / 对话模型 / 向量模型）

• 建议问题 chips，点击自动填入并发送

• 深色/浅色切换（localStorage 持久化）

• 清空对话

主聊天区：

• 用户气泡：右对齐，紫色渐变背景

• 助手气泡：左对齐，白色卡片，带渐变头像

• 流式阶段：逐 token 显示文本 + 闪烁光标，done后整体渲染 markdown

• 检索中状态：顶部状态栏显示"正在检索知识库 · search_knowledge_base"

markdown 渲染：流式阶段用renderMdLive实时渲染，自动给未闭合的代码块补上```防止布局抖动；支持标题（h1-h3）、粗体、行内代码、代码块（带语言标注）、有序/无序列表、引用、表格、链接（仅允许 http(s) 或相对路径，其余降至#）。

表格：检测到|...|行时缓冲，遇到非表格行 flush——自动识别表头行 + 分隔行（|---|）+ 数据行，CSS 带斑马纹。

参考来源：用原生<details>元素折叠，每条显示文件名、相关度分数、进度条（分数 × 100%）、180 字片段预览和"展开全文"切换。

深色模式：CSS 自定义属性 +data-theme切换，两套完整的颜色变量。

响应式：窄于 860px 时侧边栏变为抽屉面板（汉堡按钮 + 半透明遮罩）。

整个前端一个文件，约 500 行 CSS + 300 行 JS。

七、技术要点汇总

八、跑起来

# 1. 环境
python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env # 填入 DEEPSEEK_API_KEY、DASHSCOPE_API_KEY

# 2. 灌库
python ingest.py --reset

# 3. 启动
python server.py
# → http://localhost:8000

打开浏览器即可使用：提问、看流式回答、展开来源引用、切换主题。

九、接下来可以做什么

这个版本已经完整跑通 RAG + ReAct 范式，以下是可以继续增强的方向：

• 查询改写：用户口语化问题先让 LLM 重写成更适合检索的关键词组，命中率会明显提升

• Reranker：检索 Top-5 后用小模型重排（如 bge-reranker-v2），答案质量上一档

• 流式工具结果：把工具返回从一次性的整段文本改成边检索边输出的细粒度流

• 多轮对话压缩：长对话超出 token 上限时，用 sliding window 或自动摘要裁剪上下文

• 权限分级：把 BYPASS 拆成只读 / 可写的两套权限，写操作保留人工确认

• 多知识库：把不同领域的文档放入独立知识库，让 Agent 根据问题自动路由到对应知识库

这些扩展点都在 AgentScope 2.0.4 的原生支持范围内，不需要改框架就能做到。

技术栈：AgentScope 2.0.4 / DeepSeek-v4-flash / DashScope text-embedding-v3 / Qdrant (本地嵌入式) / Python 3.13 / 零额外 UI 依赖

代码量：~600 行 Python + ~800 行 HTML/CSS/JS