LlamaIndex 多文件智能问答（推荐）

概述

该实现展示如何使用 LlamaIndex 构建一个基于多文档的智能问答系统，采用新版 workflow 异步 API，结构清晰、易扩展。

技术架构

配置管理

@dataclass(frozen=True)
class AppConfig:
    docs_dir: str = "./docs"
    persist_dir: str = "./storage"
    top_k: int = 5
    query: str = "介绍下雇主责任险"
    llm_model: str = "deepseek-v3"
    embed_model: str = "text-embedding-v1"
    temperature: float = 0.7
    top_p: float = 0.8
    show_retrieval: bool = True

使用 dataclass 集中管理所有配置参数，提供类型提示和默认值，便于维护和扩展。

LLM 和 Embedding 初始化

def build_llm_and_embedding(cfg: AppConfig):
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")

    llm = SafeDashScope(
        model=cfg.llm_model,
        api_key=api_key,
        temperature=cfg.temperature,
        top_p=cfg.top_p,
    )
    embed_model = DashScopeEmbedding(
        model_name=cfg.embed_model,
        api_key=api_key,
    )
    return llm, embed_model

SafeDashScope 是对 DashScope 的封装，避免返回原始 API 响应对象，确保与 Pydantic 的兼容性。

索引构建与持久化

def build_or_load_index(cfg: AppConfig):
    if os.path.exists(cfg.persist_dir):
        try:
            storage_context = StorageContext.from_defaults(
                persist_dir=cfg.persist_dir
            )
            index = load_index_from_storage(storage_context)
            print("从存储加载索引成功")
            return index
        except Exception as exc:
            print(f"加载索引失败: {exc}，将重新创建索引")

    if not os.path.exists(cfg.docs_dir):
        print(f"文档目录 {cfg.docs_dir} 不存在")
        return None

    documents = SimpleDirectoryReader(cfg.docs_dir).load_data()
    if not documents:
        print("没有找到任何文档")
        return None

    print(f"加载了 {len(documents)} 个文档")
    index = VectorStoreIndex.from_documents(documents)
    index.storage_context.persist(persist_dir=cfg.persist_dir)
    print(f"索引已保存到 {cfg.persist_dir}")
    return index

实现索引的持久化存储，首次运行时创建索引并保存，后续运行直接加载，大幅提升启动速度。

ReAct 智能体构建

def build_agent(index, llm, cfg: AppConfig):
    query_engine = index.as_query_engine(similarity_top_k=cfg.top_k)

    def retrieve_documents(query: str) -> str:
        return str(query_engine.query(query))

    tool = FunctionTool.from_defaults(fn=retrieve_documents)

    system_prompt = (
        "你是一个乐于助人的AI助手。\n"
        "你可以从给定的文档中检索相关信息来回答用户的问题。\n"
        "你总是用中文回复用户。"
    )

    agent = ReActAgent(
        tools=[tool],
        llm=llm,
        system_prompt=system_prompt,
        verbose=True,
        streaming=False,
    )
    return agent

将检索功能封装为工具，创建 ReAct 智能体。ReAct 框架允许智能体通过推理（Reasoning）和行动（Acting）循环解决复杂问题。

异步查询执行

async def run_query(agent, index, cfg: AppConfig):
    if cfg.show_retrieval:
        retriever = index.as_retriever(similarity_top_k=cfg.top_k)
        print("\n===== 召回的文档内容 =====")
        nodes = retriever.retrieve(cfg.query)
        if nodes:
            for i, node in enumerate(nodes):
                print(f"\n文档片段 {i + 1}:")
                print(f"内容: {node.text[:200]}...")
                print(f"元数据: {node.metadata}")
                if hasattr(node, "score"):
                    print(f"相似度分数: {node.score}")
        else:
            print("没有召回任何文档内容")
        print("===========================\n")

    print("\n===== 智能体回复 =====")
    result = await agent.run(cfg.query)
    if hasattr(result, "response") and hasattr(result.response, "content"):
        print(result.response.content)
    else:
        print(result)
    print("======================\n")

使用 async/await 实现异步执行，提升性能。可选显示召回的文档片段，便于调试和理解检索过程。

命令行参数支持

核心优势

1. 异步架构

全程使用 async/await，适配 LlamaIndex 新版 workflow API，性能更优，适合高并发场景。

2. 索引持久化

首次运行创建索引并保存到本地，后续运行直接加载，避免重复计算，启动速度快。

3. ReAct 框架

智能体可以进行多步推理，根据需要调用检索工具，处理复杂查询更加智能。

4. 灵活配置

支持命令行参数和配置类，易于调整模型、目录、TopK 等参数，适应不同场景。

应用场景

这种实现方式特别适合：

• 需要高性能的生产环境
• 文档量大、需要持久化索引的场景
• 需要灵活配置的企业应用
• 需要详细日志和调试信息的开发阶段
• 需要异步处理的高并发系统

扩展建议

可以进一步优化：

• 添加多种检索策略（混合检索、重排序等）
• 实现流式输出，提升用户体验
• 添加对话历史管理，支持多轮对话
• 集成更多工具（搜索、计算器等）
• 添加评估指标，监控系统性能