从 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 个核心文件：

模块	文件	职责
配置层	config.py	读 .env，构造对话模型、嵌入模型、向量库
知识库	ingest.py	把 5 篇技术文章切片、嵌入、写入 Qdrant
智能体	agent.py	把检索能力封装成 FunctionTool，挂到 ReAct Agent
UI	server.py  + static/index.html	Python 标准库 SSE 后端 + 自建前端

四个文件不到 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 标准库 + 自建前端 的组合：

层	选型	说明
后端	http.server.ThreadingHTTPServer	Python 自带，零额外依赖
协议	Server-Sent Events（SSE）	单向流，比 WebSocket 简单；前端用 fetch + ReadableStream 直接消费
前端	单文件 index.html	内联 CSS/JS，零构建步骤

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、httpx AsyncStream）维护的异步任务生命周期跨越整个 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	DeepSeek 无 embedding 接口	嵌入层选 DashScope（或 Ollama 本地）
2	Qdrant path 模式有文件锁	避免多进程同时访问同一目录
3	chunk.content  是 TextBlock 列表	取文本用 block_text() 做类型判断
4	ReAct 默认权限会暂停等待确认	只读工具设 PermissionMode.BYPASS
5	流式事件 type 是大写字符串	比对前先 .lower()
6	AgentScope 流式依赖持久事件循环	后台 daemon 线程 + run_coroutine_threadsafe
7	/api/stats  不能在已运行的 loop 上调 run_until_complete	统一用 run_coroutine_threadsafe
8	服务器绑 127.0.0.1 可能被 IPv6 跳过	绑 0.0.0.0
9	用错 Python 解释器会导致包找不到	始终用 venv 的 python.exe

八、跑起来

# 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