Research note / 2026/07/07
项目实战-智能文档助手
Agent应用开发 · Day 7 学习笔记
日期：2026-06-09
主题：综合实战 — 从零搭建一个完整 Agent 应用
进度：40/40 概念（全部掌握）
实战项目：智能文档助手
一个能读取用户上传的文档、回答问题、执行相关操作的 Agent
整合能力清单
- ✅ 系统提示词设计（Day 2）
- ✅ ReAct 架构（Day 3）
- ✅ Function Calling + 多工具调用（Day 4）
- ✅ RAG 文档检索（Day 4）
- ✅ 对话记忆管理（Day 5）
- ✅ 错误处理与重试（Day 4）
- ✅ 基本的可观测性（Day 6）
一、需求分析与架构设计
功能需求
用户上传文档（PDF/TXT/MD）
→ Agent 解析并索引文档
→ 用户可以对文档提问
→ Agent 基于文档内容回答
→ Agent 可以执行额外操作（搜索、计算、总结）
→ 支持多轮对话，记住上下文技术架构
┌─────────────────────────────────────────────────┐
│ 前端 (Streamlit) │
│ 文件上传 │ 对话界面 │ 中间状态展示 │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ Agent 核心 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 记忆管理 │ │ ReAct循环 │ │ Trace记录│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │
│ ┌────▼────┐ ┌────▼────┐ │
│ │ 对话历史 │ │ 工具调度 │ │
│ │ 长期记忆 │ │ RAG检索 │ │
│ └─────────┘ └─────────┘ │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 工具层 │
│ 文档检索 │ 网页搜索 │ 计算器 │ 文本摘要 │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 存储层 │
│ ChromaDB (向量) │ 文件系统 │ JSON (记忆) │
└─────────────────────────────────────────────────┘技术选型
组件选择理由LLMOpenAI GPT-4oFunction Calling 支持好向量数据库ChromaDB轻量、Python 原型友好EmbeddingOpenAI text-embedding-3-small性价比高文档解析PyPDF2 + 原生读取覆盖 PDF/TXT/MD前端Streamlit快速搭建、Python 全栈框架不依赖框架，原生实现学习原理
二、项目结构
doc-assistant/
├── app.py # Streamlit 前端入口
├── agent/
│ ├── __init__.py
│ ├── core.py # Agent 核心（ReAct 循环）
│ ├── memory.py # 记忆管理
│ └── tracer.py # Trace 记录
├── tools/
│ ├── __init__.py
│ ├── registry.py # 工具注册表
│ ├── doc_search.py # 文档检索工具
│ ├── web_search.py # 网页搜索工具
│ ├── calculator.py # 计算器工具
│ └── summarizer.py # 文本摘要工具
├── rag/
│ ├── __init__.py
│ ├── chunker.py # 文档分块
│ ├── embedder.py # Embedding 封装
│ └── retriever.py # 向量检索
├── config.py # 配置管理
├── requirements.txt
└── .env # API Key三、核心模块实现
3.1 配置管理
# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
@dataclass(frozen=True)
class Config:
openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
model: str = "gpt-4o"
embedding_model: str = "text-embedding-3-small"
max_agent_steps: int = 5
chunk_size: int = 500
chunk_overlap: int = 50
top_k_results: int = 3
max_history_turns: int = 10
def validate(self):
if not self.openai_api_key:
raise ValueError("OPENAI_API_KEY 未配置")
config = Config()
config.validate()3.2 工具注册表
# tools/registry.py
from typing import Callable, Dict, Any
class Tool:
def __init__(self, name: str, description: str, func: Callable, parameters: dict):
self.name = name
self.description = description
self.func = func
self.parameters = parameters
def to_schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
def execute(self, **kwargs) -> str:
try:
return self.func(**kwargs)
except Exception as e:
return f"工具执行失败：{str(e)}"
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
def get(self, name: str) -> Tool | None:
return self._tools.get(name)
def get_all_schemas(self) -> list:
return [t.to_schema() for t in self._tools.values()]
def get_all_names(self) -> list:
return list(self._tools.keys())
# 全局注册表
registry = ToolRegistry()3.3 文档检索工具
# tools/doc_search.py
from tools.registry import Tool, registry
class DocSearchTool:
def __init__(self, retriever):
self.retriever = retriever
def search(self, query: str) -> str:
results = self.retriever.retrieve(query, top_k=3)
if not results:
return "未找到相关文档内容。"
return "\n\n".join([f"[文档片段 {i+1}]\n{r}" for i, r in enumerate(results)])
# 注册
doc_search = DocSearchTool(retriever)
registry.register(Tool(
name="search_document",
description="在用户上传的文档中搜索相关信息。当用户询问文档相关问题时使用。",
func=doc_search.search,
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询，用自然语言描述要找的内容"
}
},
"required": ["query"]
}
))3.4 RAG 检索器
# rag/retriever.py
import chromadb
from openai import OpenAI
from config import config
class Retriever:
def __init__(self):
self.client = chromadb.Client()
self.collection = self.client.create_collection("documents")
self.openai = OpenAI(api_key=config.openai_api_key)
def add_documents(self, chunks: list[str], metadatas: list[dict]):
embeddings = self._get_embeddings(chunks)
ids = [f"doc_{i}" for i in range(len(chunks))]
self.collection.add(
documents=chunks,
embeddings=embeddings,
metadatas=metadatas,
ids=ids
)
def retrieve(self, query: str, top_k: int = 3) -> list[str]:
query_embedding = self._get_embeddings([query])[0]
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
return results["documents"][0] if results["documents"] else []
def _get_embeddings(self, texts: list[str]) -> list[list[float]]:
response = self.openai.embeddings.create(
model=config.embedding_model,
input=texts
)
return [item.embedding for item in response.data]3.5 文档分块器
# rag/chunker.py
from config import config
class Chunker:
def __init__(self, chunk_size: int = None, overlap: int = None):
self.chunk_size = chunk_size or config.chunk_size
self.overlap = overlap or config.chunk_overlap
def chunk(self, text: str, source: str = "unknown") -> list[dict]:
# 递归分块：先按段落，再按句子，最后按字符
chunks = self._recursive_split(text)
return [
{
"text": chunk,
"metadata": {"source": source, "index": i}
}
for i, chunk in enumerate(chunks)
]
def _recursive_split(self, text: str) -> list[str]:
if len(text) <= self.chunk_size:
return [text.strip()] if text.strip() else []
# 尝试按分隔符切分
for separator in ["\n\n", "\n", "。", ".", " "]:
if separator in text:
parts = text.split(separator)
chunks = []
current = ""
for part in parts:
if len(current) + len(part) + len(separator) <= self.chunk_size:
current += part + separator
else:
if current.strip():
chunks.append(current.strip())
current = part + separator
if current.strip():
chunks.append(current.strip())
return chunks
# 兜底：按字符切分
return [text[i:i+self.chunk_size] for i in range(0, len(text), self.chunk_size - self.overlap)]3.6 记忆管理
# agent/memory.py
import json
from pathlib import Path
from openai import OpenAI
from config import config
class MemoryManager:
def __init__(self, user_id: str = "default"):
self.user_id = user_id
self.short_term: list[dict] = [] # 对话历史
self.long_term: list[dict] = [] # 持久化记忆
self.openai = OpenAI(api_key=config.openai_api_key)
self._load_long_term()
def add_message(self, role: str, content: str):
self.short_term.append({"role": role, "content": content})
# 自动压缩
if len(self.short_term) > config.max_history_turns * 2:
self._compress_history()
def get_history(self) -> list[dict]:
return self.short_term
def get_context(self, user_input: str) -> str:
# 检索相关长期记忆
relevant = self._recall(user_input)
if relevant:
return "用户相关信息：\n" + "\n".join(f"- {m}" for m in relevant)
return ""
def save_to_long_term(self, content: str):
self.long_term.append({"content": content, "user_id": self.user_id})
self._persist_long_term()
def _compress_history(self):
old = self.short_term[:-6]
recent = self.short_term[-6:]
summary = self.openai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "将以下对话压缩为简短摘要，保留关键信息。"},
{"role": "user", "content": json.dumps(old, ensure_ascii=False)}
],
max_tokens=300
)
self.short_term = [
{"role": "system", "content": f"对话摘要：{summary.choices[0].message.content}"}
] + recent
def _recall(self, query: str) -> list[str]:
# 简单的关键词匹配（生产环境用向量检索）
return [m["content"] for m in self.long_term if self.user_id == m.get("user_id")][:3]
def _load_long_term(self):
path = Path(f"data/memory_{self.user_id}.json")
if path.exists():
self.long_term = json.loads(path.read_text())
def _persist_long_term(self):
path = Path(f"data/memory_{self.user_id}.json")
path.parent.mkdir(exist_ok=True)
path.write_text(json.dumps(self.long_term, ensure_ascii=False, indent=2))3.7 Trace 记录
# agent/tracer.py
import time
import json
from pathlib import Path
from contextlib import contextmanager
class Tracer:
def __init__(self, trace_id: str):
self.trace_id = trace_id
self.steps: list[dict] = []
self.start_time = time.time()
self.total_tokens = 0
@contextmanager
def step(self, step_type: str, **kwargs):
step_data = {"type": step_type, "start": time.time(), **kwargs}
try:
yield step_data
except Exception as e:
step_data["error"] = str(e)
raise
finally:
step_data["duration_ms"] = round((time.time() - step_data["start"]) * 1000)
self.steps.append(step_data)
def add_tokens(self, count: int):
self.total_tokens += count
def save(self):
trace = {
"trace_id": self.trace_id,
"total_duration_ms": round((time.time() - self.start_time) * 1000),
"total_tokens": self.total_tokens,
"steps": self.steps
}
path = Path(f"traces/{self.trace_id}.json")
path.parent.mkdir(exist_ok=True)
path.write_text(json.dumps(trace, ensure_ascii=False, indent=2))3.8 Agent 核心（ReAct 循环）
# agent/core.py
import json
import uuid
from openai import OpenAI
from config import config
from tools.registry import registry
from agent.memory import MemoryManager
from agent.tracer import Tracer
class DocumentAgent:
def __init__(self, user_id: str = "default"):
self.openai = OpenAI(api_key=config.openai_api_key)
self.memory = MemoryManager(user_id)
self.tracer = None
def chat(self, user_input: str) -> str:
self.tracer = Tracer(trace_id=str(uuid.uuid4())[:8])
self.memory.add_message("user", user_input)
# 获取用户相关的长期记忆
memory_context = self.memory.get_context(user_input)
# 构建消息
messages = self._build_messages(user_input, memory_context)
# ReAct 循环
response = self._react_loop(messages)
self.memory.add_message("assistant", response)
self.tracer.save()
return response
def _build_messages(self, user_input: str, memory_context: str) -> list[dict]:
system_prompt = """你是一个智能文档助手。你可以：
1. 基于用户上传的文档回答问题（使用 search_document 工具）
2. 进行网页搜索补充信息
3. 执行数学计算
4. 对长文本进行摘要
行为规则：
- 先搜索文档，再回答。不要编造文档中没有的信息。
- 如果文档中找不到答案，如实告知用户。
- 回答简洁准确，引用文档原文时标注来源。
请使用 Thought → Action → Observation 的方式思考和执行。"""
if memory_context:
system_prompt += f"\n\n{memory_context}"
messages = [{"role": "system", "content": system_prompt}]
messages.extend(self.memory.get_history())
return messages
def _react_loop(self, messages: list[dict]) -> str:
for step in range(config.max_agent_steps):
with self.tracer.step("llm_call", step=step) as s:
response = self.openai.chat.completions.create(
model=config.model,
messages=messages,
tools=registry.get_all_schemas(),
tool_choice="auto"
)
message = response.choices[0].message
self.tracer.add_tokens(response.usage.total_tokens)
s["tokens"] = response.usage.total_tokens
# 没有工具调用，直接返回文本
if not message.tool_calls:
return message.content
# 处理工具调用
messages.append(message)
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
with self.tracer.step("tool_call", tool=func_name, args=func_args) as s:
tool = registry.get(func_name)
if tool:
result = tool.execute(**func_args)
else:
result = f"错误：工具 '{func_name}' 不存在"
s["result"] = result[:200]
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
return "抱歉，我尝试了多种方式但无法完成任务。请尝试简化您的问题。"四、前端界面
# app.py
import streamlit as st
from pathlib import Path
from agent.core import DocumentAgent
from rag.chunker import Chunker
from rag.retriever import Retriever
st.set_page_config(page_title="智能文档助手", layout="wide")
st.title("📄 智能文档助手")
# 初始化
if "agent" not in st.session_state:
st.session_state.agent = DocumentAgent()
st.session_state.retriever = Retriever()
st.session_state.chunker = Chunker()
# 侧边栏：文档上传
with st.sidebar:
st.header("📁 上传文档")
uploaded_file = st.file_uploader("选择文件", type=["pdf", "txt", "md"])
if uploaded_file:
with st.spinner("正在处理文档..."):
# 读取文件内容
if uploaded_file.type == "application/pdf":
import PyPDF2
reader = PyPDF2.PdfReader(uploaded_file)
content = "\n".join(page.extract_text() for page in reader.pages)
else:
content = uploaded_file.read().decode("utf-8")
# 分块并索引
chunks = st.session_state.chunker.chunk(content, source=uploaded_file.name)
st.session_state.retriever.add_documents(
chunks=[c["text"] for c in chunks],
metadatas=[c["metadata"] for c in chunks]
)
st.success(f"文档已索引：{len(chunks)} 个片段")
# 对话界面
for msg in st.session_state.get("messages", []):
with st.chat_message(msg["role"]):
st.write(msg["content"])
user_input = st.chat_input("输入你的问题...")
if user_input:
st.session_state.messages = st.session_state.get("messages", [])
st.session_state.messages.append({"role": "user", "content": user_input})
with st.chat_message("user"):
st.write(user_input)
with st.chat_message("assistant"):
with st.spinner("思考中..."):
response = st.session_state.agent.chat(user_input)
st.write(response)
st.session_state.messages.append({"role": "assistant", "content": response})五、运行与测试
安装依赖
pip install openai chromadb streamlit PyPDF2 python-dotenv配置环境变量
# .env
OPENAI_API_KEY=sk-your-key-here启动应用
streamlit run app.py测试用例
测试1：文档问答
上传一篇技术文档 → 提问"这篇文章的主要观点是什么？"
预期：基于文档内容回答，不编造
测试2：多轮对话
问"文档中提到了哪些技术？" → 再问"这些技术的优缺点是什么？"
预期：第二轮能理解"这些技术"指代第一轮的结果
测试3：工具选择
问"123 * 456 等于多少？"
预期：调用计算器工具，不自己算
测试4：文档外问题
问"今天天气怎么样？"（文档中没有天气信息）
预期：告知文档中没有相关信息六、Day 7 核心收获
从 7 天学习中提炼的 Agent 开发心法
- Agent = LLM + 循环 + 工具：这是最小定义，缺一不可
- Prompt 是最核心的工程：80% 的问题可以通过改进 Prompt 解决
- 工具描述决定工具选择：花时间写好 description，比调参数有用
- 错误处理比正常流程更重要：Agent 的非决定性意味着什么错误都可能发生
- 记忆是 Agent 的灵魂：没有记忆的 Agent 只是高级 chatbot
- RAG 是 Agent 的知识库：让 Agent 能访问私有数据的关键技术
- Trace 是调试的生命线：没有 Trace，出了问题你都不知道发生了什么
- 安全是底线：Prompt 注入、权限管控、危险操作确认，一个都不能少
- 成本控制是生存线：Token 就是钱，精打细算才能持续运行
- 迭代是唯一的方法论：没有一步到位的 Agent，只有不断改进的 Agent
7 天知识图谱回顾
Day 1: 认知地基 Agent 是什么 + LLM 基础
Day 2: 对话的艺术 Prompt Engineering
Day 3: 思维方式 Agent 架构模式（ReAct / Plan / Reflexion）
Day 4: 手和眼 工具使用 + RAG
Day 5: 记忆与协作 记忆系统 + 多 Agent
Day 6: 质量与上线 评估调试 + 部署工程化
Day 7: 综合实战 从零搭建完整 Agent ← 你在这里下一步方向
方向学习内容深入 RAG混合检索、重排序、多模态 RAG生产部署Docker、K8s、监控告警高级多 AgentAutoGen、CrewAI、复杂协作模式评估体系自动化评估流水线、LLM-as-Judge垂直领域代码 Agent、数据分析 Agent、客服 Agent
Discussion
评论
站点尚未配置 GitHub Discussions。文章阅读不受影响，也不会再发起无效的 GitHub 登录请求。