Agent 基础：

智能体（Agent）是一种让大语言模型（LLM）具备自主规划、决策与执行能力的组件，核心是让模型根据任务需求，主动选择并调用外部工具，解决单靠模型自身无法完成的复杂问题

没有 Agent 时：LLM 只能依赖自身训练数据回答问题，遇到需要实时数据、复杂计算、外部资源调用的场景，就会因能力边界而 "卡壳"
有了 Agent 后：LLM 就像一个 "指挥官"，会形成闭环工作流：
思考任务步骤 → 选择合适工具 → 执行工具调用 → 根据结果调整策略 → 循环迭代直到完成任务

整体工作流程：

用户输入：用户提出任务需求，传递给 Agent 智能体

Agent 中枢处理：智能体接收需求后，结合三大核心模块进行闭环决策：

工具：连接外部系统的接口，弥补 LLM 的局限性（如调用数据库、搜索引擎、API 等）

记忆：存储对话历史与任务上下文，支持多轮对话与持续任务处理

规划：将复杂任务拆解为可执行的步骤，自主决策下一步行动

迭代执行：根据工具返回的结果，Agent 自主判断是否继续调用工具，还是生成最终答案

回复用户：将整理好的最终结果返回给用户

普通 Chain 流程固定、工具调用路径写死，只适合简单标准化任务；而 Agent 流程动态可变，工具选择由大模型自主决策，更适合复杂、多步骤、需要决策的任务

工具调用机制：

外部工具赋予了大模型感知和影响现实世界的能力，极大地突破了它的能力边界；丰富的工具集能显著提升大模型的工作性能，拓展其业务范畴；工具越多，Agent 能覆盖的场景就越广，从客服问答到库存管理再到自动化运营，实用性也随之大幅提升

设计一个天气查询工具，让大模型具有天气问答的能力：

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

# 使用 @tool 装饰器把普通函数注册成一个工具
@tool(description="查询天气")
def get_weather() -> str:
    return "晴天"

# 创建一个智能体 agent
agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    system_prompt="你是一个聊天助手, 可以回答用户问题",
)

res = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "明天深圳的天气如何？"}
        ]
    }
)

for msg in res["messages"]:
    print(type(msg).__name__, msg.content)

流式输出机制：

在 LangChain 中，通过 create_agent 创建的 Agent 对象，本质上是 Runnable 接口的实现，因此支持和普通 Chain 一致的执行方式；其中，stream 方法就是用来实现流式输出的：它会在 Agent 执行过程中，将思考过程、工具调用和模型生成的内容，分段实时返回，而不是等全部处理完再一次性给出完整结果

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取股价, 传入股票名称, 返回字符串信息")
def get_price(name: str) -> str:
    return f"股票{name}的价格是20元"


@tool(description="获取股票信息, 传入股票名称, 返回字符串信息")
def get_info(name: str) -> str:
    return f"股票{name}，是一家A股上市公司, 专注于IT职业教育"

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_price, get_info],
    system_prompt="你是一个智能助手, 可以回答股票相关问题, 记住请告知我思考过程, 让我知道你为什么调用某个工具"
)

# 以流式方式调用智能体并逐步获取结果
for chunk in agent.stream(
        {"messages": [{"role": "user", "content": "爱坤教育股价多少, 并介绍一下"}]},
        stream_mode="values"  # 每次返回当前完整的状态数据
):
    # 取出当前返回结果中的最后一条消息
    latest_message = chunk["messages"][-1]

    # 如果这条消息有内容，就打印消息类型和内容
    if latest_message.content:
        print(type(latest_message).__name__, latest_message.content)

    try:
        # 如果这条消息包含工具调用信息，就打印工具名称
        if latest_message.tool_calls:
            print(f"工具调用: {[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        # 某些消息对象没有 tool_calls 属性时，忽略异常
        pass

ReAct 行动框架：

Agent ReAct（全称 Reasoning + Acting，即 "推理 + 行动"）是大模型智能体的核心思考与行动框架，它让 Agent 模仿人类解决问题的逻辑：先思考问题、制定策略，再执行行动、验证结果，形成闭环迭代

简单来说，ReAct 改变了大模型 "直接输出答案" 的模式，而是通过自然语言思考过程来指导工具调用，一步步解决复杂问题，非常适配多步推理、工具协作的场景（比如智能客服、报告生成、任务规划等）

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool


@tool(description="获取体重，返回值是整数，单位千克")
def get_weight() -> int:
    return 55


@tool(description="获取身高，返回值是整数，单位厘米")
def get_height() -> int:
    return 173


agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weight, get_height],
    system_prompt="""你是严格遵循ReAct框架的智能体，必须按「思考→行动→观察→再思考」的流程解决问题，
    且**每轮仅能思考并调用1个工具**，禁止单次调用多个工具；
    并告知我你的思考过程，工具的调用原因，按思考、行动、观察三个结构告知我""",
)

# 以流式方式调用智能体并逐步获取结果
for chunk in agent.stream(
        {"messages": [{"role": "user", "content": "计算我的BMI"}]},
        stream_mode="values"  # 每次返回当前完整的状态数据
):
    # 取出当前返回结果中的最后一条消息
    latest_message = chunk["messages"][-1]

    # 如果这条消息有内容, 就打印消息类型和内容
    if latest_message.content:
        print(type(latest_message).__name__, latest_message.content)

    try:
        # 如果这条消息包含工具调用信息, 就打印工具名称
        if latest_message.tool_calls:
            print(f"工具调用: {[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        # 某些消息对象没有 tool_calls 属性时, 忽略异常
        pass

Middleware 中间件机制：

Agent 中间件是对智能体每一步执行过程进行拦截和自定义控制的扩展机制，通过 Hooks（钩子）在关键节点插入自定义逻辑，实现日志记录、调试、重试、安全防护等功能

无中间件时：Agent 流程是简单的 request → model → tools/result

加入中间件后：可以在 before_agent、before_model、wrap_model_call、wrap_tool_call、after_model、after_agent 等多个节点进行拦截，实现对模型调用、工具调用的精细化控制

这些钩子支持用装饰器快速定义，LangChain 也提供了内置的基础中间件，让开发者可以轻松扩展 Agent 的能力

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent, after_agent, before_model, after_model, wrap_model_call, \
    wrap_tool_call
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langgraph.runtime import Runtime


@tool(description="查询天气，传入城市名称字符串，返回字符串天气信息")
def get_weather(city: str) -> str:
    return f"{city}天气：晴天"

# 在 agent 开始执行之前注册一个钩子函数
# 只要 agent 一启动，这个函数就会被自动调用
@before_agent
def log_before_agent(state: AgentState, runtime: Runtime) -> None:
    # state 表示当前 agent 的状态数据，里面通常包含 messages 等信息
    # runtime 表示当前运行时上下文，里面可能包含本次运行的环境信息
    print(f"[before agent]agent启动，并附带{len(state['messages'])}消息")


# 在 agent 执行结束之后注册一个钩子函数
# 当 agent 全部执行完成后，这个函数会被自动调用
@after_agent
def log_after_agent(state: AgentState, runtime: Runtime) -> None:
    # 这里打印 agent 结束时的消息数量
    # 通常结束时的消息数会比开始时更多，因为中间可能新增了 AI 回复或工具消息
    print(f"[after agent]agent结束，并附带{len(state['messages'])}消息")


# 在模型调用之前注册一个钩子函数
# 每次 agent 准备调用大模型前，都会先执行这个函数
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> None:
    # 打印模型调用前状态中的消息数量
    # 这有助于观察模型在接收多少上下文消息后开始工作
    print(f"[before_model]模型即将调用，并附带{len(state['messages'])}消息")


# 在模型调用完成之后注册一个钩子函数
# 每次模型返回结果后，这个函数都会被自动调用
@after_model
def log_after_model(state: AgentState, runtime: Runtime) -> None:
    # 打印模型调用结束后的消息数量
    # 通过和 before_model 对比，可以观察模型调用后状态是否发生了变化
    print(f"[after_model]模型调用结束，并附带{len(state['messages'])}消息")


# 注册一个"包裹模型调用"的钩子函数
# 这个函数会包在真正的模型调用外面，所以你可以在这里做监控或日志
@wrap_model_call
def model_call_hook(request, handler):
    # 在真正调用模型前打印一条日志
    print("模型调用了")
    # handler(request) 表示继续执行真正的模型调用
    # 如果不写这句，模型就不会真的被调用
    return handler(request)


# 注册一个"包裹工具调用"的钩子函数
# 每次 agent 决定调用工具时，都会先进入这个函数
@wrap_tool_call
def monitor_tool(request, handler):
    # request.tool_call 表示当前这次工具调用的信息
    # ['name'] 表示工具名称，比如 get_weather
    print(f"工具执行：{request.tool_call['name']}")

    # ['args'] 表示本次调用这个工具时传入的参数
    print(f"工具执行传入参数：{request.tool_call['args']}")

    # handler(request) 表示继续执行真正的工具调用
    # 如果不写这句，工具只会打印日志，不会真的运行
    return handler(request)


agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    middleware=[log_before_agent, log_after_agent, log_before_model, log_after_model, model_call_hook, monitor_tool]
)

res = agent.invoke({"messages": [{"role": "user", "content": "深圳今天的天气如何呀，如何穿衣"}]})
print(res)

Agent 入门项目：

项目简介：

智扫通 Agent 项目是面向消费者的扫地机器人智能客服系统，为用户提供全周期服务，核心功能分为两部分：

智能问答服务：覆盖用户购买前的产品咨询（功能、价格、对比）与购买后的使用问题（操作指导、故障处理、维护建议），通过 RAG 技术检索知识库信息，提供及时可靠的回答
使用报告与优化建议生成：针对已购用户，自动分析设备的使用数据（清洁频率、耗材状态、错误日志等），生成个性化报告并提供优化建议（清洁计划调整、部件更换提醒），支持用户主动查询或系统定期推送

项目结构：

项目以 react_agent.py 中的 ReactAgent 类为核心，整合了三大部分：

工具与中间件：agent_tools.py 提供了问答、数据获取等业务工具，middleware.py 实现日志、监控等拦截逻辑
RAG 服务：rag_service.py 封装了问答链，依赖 vector_store.py 实现知识库检索，由 factory.py 提供模型实例
通用工具与入口：utils 包提供配置、文件、日志等基础工具，最终由 app.py 作为 Streamlit 网页入口对外提供服务

基础工具模块：

日志和路径工具：

path_tools：

import os

def get_project_root() -> str:
    # 获取当前文件的绝对路径
    current_file = os.path.abspath(__file__)
    # 获取当前文件所在目录
    current_dir = os.path.dirname(current_file)
    # 当前目录的上一级目录作为项目根目录
    project_root = os.path.dirname(current_dir)
    return project_root

def get_abs_path(relative_path: str) -> str:
    # 基于项目根目录拼接相对路径，返回绝对路径
    project_root = get_project_root()
    return os.path.join(project_root, relative_path)

logger_handler：

import logging
import os
from datetime import datetime
from utils.path_tools import get_abs_path

# 日志保存根目录
LOG_ROOT = get_abs_path("logs")
# 确保日志目录存在
os.makedirs(LOG_ROOT, exist_ok=True)

# 定义统一的日志输出格式
# 包含：时间、logger 名称、日志级别、文件名、行号、日志消息
DEFAULT_LOG_FORMAT = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

# 获取日志器
def get_logger(
    name: str = "agent", # 日志器名字
    console_level: int = logging.INFO, # 控制台输出的最低级别
    file_level: int = logging.DEBUG, # 文件输出的最低级别
    log_file=None, # 日志文件路径
) -> logging.Logger:
    # 根据名称获取 logger；同名 logger 通常是同一个对象
    logger = logging.getLogger(name)

    # 设置 logger 的总级别为 DEBUG，表示它接收 DEBUG 及以上的日志
    # 真正输出哪些日志，还要看各个 handler 的级别
    logger.setLevel(logging.DEBUG)

    # 如果这个 logger 已经添加过 handler，就直接返回
    # 防止重复添加 handler，导致日志打印多次
    if logger.handlers:
        return logger

    # 创建控制台输出 handler
    console_handler = logging.StreamHandler()
    # 设置控制台输出的最低日志级别
    console_handler.setLevel(console_level)
    # 给控制台 handler 设置统一日志格式
    console_handler.setFormatter(DEFAULT_LOG_FORMAT)

    # 把控制台 handler 添加到 logger
    logger.addHandler(console_handler)

    # 如果没有指定日志文件路径，则自动生成一个按日期命名的日志文件
    if not log_file:
        log_file = os.path.join(
            LOG_ROOT,
            f"{name}_{datetime.now().strftime('%Y%m%d')}.log"
        )

    # 创建文件输出 handler，日志写入文件，编码为 utf-8
    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    # 设置文件输出的最低日志级别
    file_handler.setLevel(file_level)
    # 给文件 handler 设置统一日志格式
    file_handler.setFormatter(DEFAULT_LOG_FORMAT)

    # 把文件 handler 添加到 logger
    logger.addHandler(file_handler)

    # 返回配置完成的 logger
    return logger


# 创建一个默认 logger，名称为 agent
logger = get_logger()


if __name__ == "__main__":
    logger.info("信息日志")
    logger.error("错误日志")
    logger.warning("警告日志")
    # DEBUG 级别日志：默认只会写入文件，不会显示在控制台
    logger.debug("调试日志")

配置和文件工具：

config_handler：

import yaml
from utils.path_tools import get_abs_path

class ConfigHandler(object):

    @staticmethod
    def load_rag_config(config_path: str = get_abs_path("config/rag.yml"), encoding="utf-8"):
        # 读取 RAG 配置文件并解析为 Python 对象
        with open(config_path, "r", encoding=encoding) as f:
            # yaml.load(...) 把 YAML 字符串解析成 Python 对象, 通常得到的是 dict
            return yaml.load(f.read(), Loader=yaml.FullLoader)

    @staticmethod
    def load_chroma_config(config_path: str = get_abs_path("config/chroma.yml"), encoding="utf-8"):
        # 读取 Chroma 配置文件并解析为 Python 对象
        with open(config_path, "r", encoding=encoding) as f:
            return yaml.load(f.read(), Loader=yaml.FullLoader)

    @staticmethod
    def load_prompts_config(config_path: str = get_abs_path("config/prompts.yml"), encoding="utf-8"):
        # 读取提示词配置文件并解析为 Python 对象
        with open(config_path, "r", encoding=encoding) as f:
            return yaml.load(f.read(), Loader=yaml.FullLoader)

    @staticmethod
    def load_agent_config(config_path: str = get_abs_path("config/agent.yml"), encoding="utf-8"):
        # 读取 Agent 配置文件并解析为 Python 对象
        with open(config_path, "r", encoding=encoding) as f:
            return yaml.load(f.read(), Loader=yaml.FullLoader)


# 加载各模块配置，供项目其他位置直接使用
rag_conf = ConfigHandler.load_rag_config()
chroma_conf = ConfigHandler.load_chroma_config()
prompts_conf = ConfigHandler.load_prompts_config()
agent_conf = ConfigHandler.load_agent_config()

if __name__ == "__main__":
    print(rag_conf["chat_model_name"])

file_handler：

import os
import hashlib
from utils.logger_handler import logger
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_community.document_loaders import CSVLoader

# 计算指定文件的 MD5 值，并返回对应的十六进制字符串
def get_file_md5_hex(filepath: str):

    if not os.path.exists(filepath):
        # 如果路径不存在，则记录错误日志并直接返回
        logger.error(f"[md5计算]文件{filepath}不存在")
        return
    if not os.path.isfile(filepath):
        # 如果路径存在但不是普通文件，则记录错误日志并返回
        logger.error(f"[md5计算]路径{filepath}不是文件")
        return
    # 创建一个 MD5 哈希对象，用于逐步累加文件内容
    md5_obj = hashlib.md5()
    # 每次读取 4096 字节，避免一次性读取大文件占用过多内存
    chunk_size = 4096
    try:
        # 以二进制只读方式打开文件，确保哈希计算基于原始字节内容
        with open(filepath, "rb") as f:
            # 按块循环读取文件内容，并持续更新到 MD5 对象中
            while chunk := f.read(chunk_size):
                md5_obj.update(chunk)
        """
        # 与上面 while chunk := f.read(...) 等价的传统写法
        chunk = f.read(chunk_size)
        while chunk:
            md5_obj.update(chunk)
            chunk = f.read(chunk_size)
        """
        # 将最终 MD5 结果转换为十六进制字符串并返回
        md5_hex = md5_obj.hexdigest()
        return md5_hex
    except Exception as e:
        # 如果读取或计算过程中发生异常，则记录错误信息并返回 None
        logger.error(f"计算文件{filepath}md5失败，{str(e)}")
        return None

# 返回指定文件夹中所有后缀属于 allowed_types 的文件路径
def listdir_with_allowed_type(path: str, allowed_types: tuple[str]):

    files = []
    if not os.path.isdir(path):
        # 如果传入路径不是文件夹，则记录错误日志并返回空元组
        logger.error(f"[listdir_with_allowed_type]{path}不是文件夹")
        return tuple()
    for f in os.listdir(path):
        # 如果当前文件名以允许的后缀结尾，则加入结果列表
        if f.endswith(allowed_types):
            files.append(os.path.join(path, f))
    # 将结果列表转换为元组返回，便于作为只读结果使用
    return tuple(files)


def pdf_loader(filepath: str, passwd=None) -> list[Document]:
    # 使用 PyPDFLoader 加载 PDF 文件，并返回 Document 对象列表
    return PyPDFLoader(filepath, passwd).load()


def txt_loader(filepath: str) -> list[Document]:
    # 使用 TextLoader 加载文本文件，并返回 Document 对象列表
    return TextLoader(filepath, encoding="utf-8").load()

def csv_loader(filepath: str) -> list[Document]:
    # 使用 CSVLoader 加载文本文件，并返回 Document 对象列表
    return CSVLoader(filepath).load()

提示词加载工具：

prompts.yml：

rag_summarize_prompt_path: prompts/rag_summarize.txt
main_prompt_path: prompts/main_prompt.txt
report_prompt_path: prompts/report_prompt.txt

prompt_loader：

from utils.config_handler import prompts_conf
from utils.path_tools import get_abs_path
from utils.logger_handler import logger

# 从 prompts_conf 中读取主提示词文件路径，并返回对应的提示词内容
def load_system_prompts():
    try:
        system_prompt_path = get_abs_path(prompts_conf["main_prompt_path"])
    except KeyError as e:
        # 如果配置中缺少 main_prompt_path 键，则记录错误并继续抛出异常
        logger.error(f"[load_system_prompts]在yaml配置项中没有main_prompt_path配置项")
        raise e

    try:
        # 按 utf-8 编码读取主提示词文件内容并返回
        return open(system_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        # 如果提示词文件读取失败，则记录错误并继续抛出异常
        logger.error(f"[load_system_prompts]解析系统提示词出错，{str(e)}")
        raise e

# 从 prompts_conf 中读取 RAG 总结提示词文件路径，并返回对应的提示词内容
def load_rag_prompts():
    try:
        rag_prompt_path = get_abs_path(prompts_conf["rag_summarize_prompt_path"])
    except KeyError as e:
        # 如果配置中缺少 rag_summarize_prompt_path 键，则记录错误并继续抛出异常
        logger.error(f"[load_rag_prompts]在yaml配置项中没有rag_summarize_prompt_path配置项")
        raise e

    try:
        # 按 utf-8 编码读取 RAG 提示词文件内容并返回
        return open(rag_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        # 如果提示词文件读取失败，则记录错误并继续抛出异常
        logger.error(f"[load_rag_prompts]解析RAG总结提示词出错，{str(e)}")
        raise e

# 从 prompts_conf 中读取报告生成提示词文件路径，并返回对应的提示词内容
def load_report_prompts():
    try:
        report_prompt_path = get_abs_path(prompts_conf["report_prompt_path"])
    except KeyError as e:
        # 如果配置中缺少 report_prompt_path 键，则记录错误并继续抛出异常
        logger.error(f"[load_report_prompts]在yaml配置项中没有report_prompt_path配置项")
        raise e

    try:
        # 按 utf-8 编码读取报告提示词文件内容并返回
        return open(report_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        # 如果提示词文件读取失败，则记录错误并继续抛出异常
        logger.error(f"[load_report_prompts]解析报告生成提示词出错，{str(e)}")
        raise e

if __name__ == "__main__":
    print(load_system_prompts())

核心服务模块：

向量存储服务：

chroma.yml：

collection_name: agent
persist_directory: chroma_db
k: 3
data_path: data
md5_hex_store: md5.text
allow_knowledge_file_type: ["txt", "pdf", "csv"]

chunk_size: 200
chunk_overlap: 20
separators: ["\n\n", "\n", ".", "!", "?", "。", "！", "？", " ", ""]

factory：

from abc import ABC
from abc import abstractmethod
from langchain_community.chat_models.tongyi import ChatTongyi, BaseChatModel
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.embeddings import Embeddings
from typing import Optional
from utils.config_handler import rag_conf

class BaseModelFactory(ABC):
    # 定义模型工厂抽象基类，约束子类实现 generator 方法
    @abstractmethod
    def generator(self) -> Optional[Embeddings | BaseChatModel]:
        # 返回聊天模型或向量模型实例
        pass

# 聊天模型工厂类，负责创建聊天模型实例
class ChatModelFactory(BaseModelFactory):
    def generator(self) -> ChatTongyi:
        # 根据配置返回 Tongyi 聊天模型对象
        return ChatTongyi(model=rag_conf["chat_model_name"])

# 向量模型工厂类，负责创建文本嵌入模型实例
class EmbeddingsFactory(BaseModelFactory):
    def generator(self) -> Embeddings:
        # 根据配置返回 DashScope 向量模型对象
        return DashScopeEmbeddings(model=rag_conf["embedding_model_name"])

# 创建聊天模型实例，供项目其他模块直接使用
chat_model = ChatModelFactory().generator()
# 创建向量模型实例，供向量库和检索流程使用
embed_model = EmbeddingsFactory().generator()

vector_store：

from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from model.factory import embed_model
from langchain_chroma import Chroma
from utils.config_handler import chroma_conf
import os
from utils.file_handler import get_file_md5_hex, listdir_with_allowed_type, csv_loader, pdf_loader, txt_loader
from utils.logger_handler import logger
from utils.path_tools import get_abs_path

# 定义向量库服务类，负责知识文件入库和检索器创建
class VectorStoreService:
    # 初始化向量库对象和文本切分器，供后续文档导入与检索使用
    def __init__(self):
        self.vector_store = Chroma(
            # 指定当前向量库集合名称，用于区分不同知识库集合
            collection_name=chroma_conf["collection_name"],
            # 指定嵌入模型，将文本内容转换为向量后再存入向量库
            embedding_function=embed_model,
            # 指定向量库的持久化目录，使向量数据能够保存在本地磁盘
            persist_directory=get_abs_path(chroma_conf["persist_directory"]),
        )

        self.splitter = RecursiveCharacterTextSplitter(
            # 指定每个文本分块的最大长度
            chunk_size=chroma_conf["chunk_size"],
            # 指定相邻分块之间的重叠长度，以保留上下文连续性
            chunk_overlap=chroma_conf["chunk_overlap"],
            # 指定优先使用的切分符列表，例如换行符、句号等
            separators=chroma_conf["separators"],
            # 指定使用 len 函数计算文本长度
            length_function=len,
        )

    # 将底层向量库包装成检索器对象，供 RAG 问答阶段直接调用
    def get_retriever(self):
        # 设置每次检索返回最相似的前 k 条文本片段
        return self.vector_store.as_retriever(search_kwargs={"k": chroma_conf["k"]})

    # 加载知识目录下的文件，去重后切分并写入向量库
    def load_document(self):
        # 检查当前文件的 MD5 是否已经记录在本地，避免重复入库
        def check_md5_hex(md5_for_check):
            # 如果 MD5 记录文件不存在，则先创建空文件并视为未入库
            if not os.path.exists(get_abs_path(chroma_conf["md5_hex_store"])):
                open(get_abs_path(chroma_conf["md5_hex_store"]), "w", encoding="utf-8").close()
                return False
            # 打开 MD5 记录文件并逐行比对是否已有当前文件指纹
            with open(get_abs_path(chroma_conf["md5_hex_store"]), "r", encoding="utf-8") as f:
                # 遍历记录文件中的每一行 MD5 值
                for line in f.readlines():
                    # 去掉行尾换行符，保证比较时字符串一致
                    line = line.strip()
                    # 如果找到相同 MD5，说明该文件内容已处理过
                    if line == md5_for_check:
                        return True
            # 如果未找到相同 MD5，则说明该文件尚未入库
            return False

        # 将新入库文件的 MD5 追加保存到本地记录文件中
        def save_md5_hex(md5_for_save):
            # 以追加模式打开 MD5 记录文件，避免覆盖已有内容
            with open(get_abs_path(chroma_conf["md5_hex_store"]), "a", encoding="utf-8") as f:
                # 将当前文件的 MD5 写入新的一行，供下次去重检查使用
                f.write(md5_for_save + "\n")

        # 根据文件扩展名选择对应加载器，并返回 Document 列表
        def get_file_documents(read_path: str):
            if read_path.endswith("txt"):
                return txt_loader(read_path)
            elif read_path.endswith("pdf"):
                return pdf_loader(read_path)
            elif read_path.endswith("csv"):
                return csv_loader(read_path)
            else:
                # 如果文件类型不支持，则返回空列表表示无可用文档
                return []

        allowed_files_path = listdir_with_allowed_type(
            # 获取知识库数据目录的绝对路径
            get_abs_path(chroma_conf["data_path"]),
            # 传入允许加载的文件后缀列表，例如 .txt、.pdf、.csv
            tuple(chroma_conf["allow_knowledge_file_type"]),
        )

        # 遍历目录下所有符合条件的知识文件并逐个处理
        for path in allowed_files_path:
            # 计算当前文件内容的 MD5 值作为唯一指纹
            md5_hex = get_file_md5_hex(path)
            # 如果 MD5 计算失败，则记录警告并跳过该文件
            if not md5_hex:
                logger.warning(f"[加载知识库] {path} MD5计算失败，跳过")
                continue
            # 如果该文件内容已存在于知识库记录中，则直接跳过避免重复导入
            if check_md5_hex(md5_hex):
                logger.info(f"[加载知识库] {path} 内容已存在于知识库，跳过")
                continue

            try:
                # 按文件类型读取原始文件内容，并转换为 LangChain 的 Document 列表
                documents: list[Document] = get_file_documents(path)
                # 如果文件没有成功解析出有效文档内容，则记录警告并跳过
                if not documents:
                    logger.warning(f"[加载知识库] {path} 无有效文本内容，跳过")
                    continue
                # 将原始文档按配置切分为更适合向量化和检索的小片段
                split_document: list[Document] = self.splitter.split_documents(documents)
                # 如果切分后没有得到有效文本片段，则记录警告并跳过
                if not split_document:
                    logger.warning(f"[加载知识库] {path} 分片后无内容，跳过")
                    continue
                # 将切分后的文档片段写入 Chroma 向量库并自动完成向量化
                self.vector_store.add_documents(split_document)
                # 入库成功后保存该文件的 MD5，防止下次重复导入
                save_md5_hex(md5_hex)
                # 记录当前文件成功写入知识库的日志信息
                logger.info(f"[加载知识库] {path} 内容加载成功")
            except Exception as e:
                # 如果文件加载、切分或入库过程中发生异常，则记录错误并继续处理下一个文件
                logger.error(f"[加载知识库] {path} 加载失败：{str(e)}", exc_info=True)
                continue


if __name__ == "__main__":
    # 创建向量库服务实例，初始化向量库和切分器
    store = VectorStoreService()
    # 扫描知识目录并将符合条件的文档加载到向量库中
    store.load_document()
    # 获取基于当前向量库的检索器对象
    retriever = store.get_retriever()
    # 使用“迷路”作为查询词，从知识库中检索最相关的文本片段
    res = retriever.invoke("迷路")
    # 遍历检索结果中的每一个文档片段
    for r in res:
        # 打印当前检索结果的正文内容
        print(r.page_content)
        # 打印分隔线，便于区分不同检索结果
        print("-" * 20)

RAG 服务：

rag_service：

from langchain_core.documents import Document
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from rag.vector_store import VectorStoreService
from langchain_core.runnables import RunnableLambda
from model.factory import chat_model
from utils.prompt_loader import load_rag_prompts

"""
总结服务类：用户提问，搜索参考资料，将提问和参考资料提交给模型，让模型总结回复
"""

# 用于把 Prompt 打印出来，方便调试查看最终传给大模型的内容
def print_prompt(prompt):
    # 打印分隔线，便于在控制台中观察输出
    print("=" * 20)
    # 将 Prompt 对象转换成字符串后输出
    print(prompt.to_string())
    # 再打印一次分隔线，形成完整的提示词显示区域
    print("=" * 20)
    # 一定要把 prompt 返回出去，这样后面的模型才能继续接收到它
    return prompt


class RagSummarizeService(object):
    def __init__(self):
        # 初始化向量库服务对象
        self.vector_store = VectorStoreService()
        # 从向量库中获取检索器，用于后续根据用户问题检索相关文档
        self.retriever = self.vector_store.get_retriever()
        # 加载 RAG 提示词模板的原始文本
        self.prompt_text = load_rag_prompts()
        # 将字符串模板转换成 LangChain 的 PromptTemplate 对象
        self.prompt_template = PromptTemplate.from_template(self.prompt_text)
        # 指定要调用的大语言模型
        self.model = chat_model
        # 初始化责任链：提示词模板 -> 打印提示词 -> 模型 -> 输出解析
        self.chain = self.__init__chain()

    def __init__chain(self):
        # 构建 LangChain 执行链
        # prompt_template：将 input 和 context 填充进模板
        # RunnableLambda(print_prompt)：打印最终 prompt，便于调试
        # self.model：调用大模型生成回答
        # StrOutputParser()：将模型输出解析为普通字符串
        chain = self.prompt_template | RunnableLambda(print_prompt) | self.model | StrOutputParser()
        return chain

    def retriever_docs(self, query: str) -> list[Document]:
        # 调用检索器，根据用户问题查询最相关的文档片段
        return self.retriever.invoke(query)

    def rag_summarize(self, query: str) -> str:
        # 先根据用户问题从知识库中检索相关文档
        context_docs = self.retriever_docs(query)
        # 用于存放最终拼接好的上下文内容
        context = ""
        # 计数器，用于给每条参考资料编号
        counter = 0
        # 遍历检索到的文档列表
        for doc in context_docs:
            counter += 1

            # 将每篇文档的正文内容和元数据拼接成提示词中的上下文
            # doc.page_content：文档正文
            # doc.metadata：文档来源、页码等附加信息
            context += f"[参考资料{counter}]：资料内容：{doc.page_content} | 资料元数据：{doc.metadata}"

        # 调用责任链执行
        # input：用户原始问题
        # context：从知识库检索后整理出的参考资料
        return self.chain.invoke(
            {
                "input": query,
                "context": context,
            }
        )


if __name__ == "__main__":
    # 创建 RAG 摘要服务对象
    rag = RagSummarizeService()
    # 输入问题并打印最终生成结果
    print(rag.rag_summarize("小户型适合哪些扫地机器人"))

Tools 工具：

agent.yml：

external_data_path: data/external/records.csv

agent_tools：

from langchain_core.tools import tool
from rag.rag_service import RagSummarizeService
import random, os
from utils.config_handler import agent_conf
from utils.path_tools import get_abs_path
from utils.logger_handler import logger

rag = RagSummarizeService()

user_ids = ["1001", "1002", "1003", "1004", "1005", "1006", "1007", "1008", "1009", "1010",]
month_arr = ["2025-01", "2025-02", "2025-03", "2025-04", "2025-05", "2025-06",
             "2025-07", "2025-08", "2025-09", "2025-10", "2025-11", "2025-12",]

external_data = {}

@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str) -> str:
    return rag.rag_summarize(query)

@tool(description="获取指定城市的天气，以纯字符串形式返回")
def get_weather(city: str) -> str:
    return f"城市{city}天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时内降雨概率极低"

@tool(description="获取用户所在城市名称，以纯字符串形式返回")
def get_user_location() -> str:
    return random.choice(["深圳", "合肥", "杭州"])

@tool(description="获取用户ID, 以纯字符串形式返回")
def get_user_id() -> str:
    return random.choice(user_ids)

@tool(description="获取当前月份，以纯字符串形式返回")
def get_current_month() -> str:
    return random.choice(month_arr)



# 生成并缓存外部数据，返回的数据结构为：
# {
#     "1001": {
#         "2025-01": {
#             "特征": "...",
#             "效率": "...",
#             "耗材": "...",
#             "对比": "..."
#         }
#     }
# }
def generate_external_data() -> dict:
    # 如果 external_data 为空，说明数据还没有加载过
    # 只有第一次调用时才会真正读取文件，后续直接复用内存中的数据
    if not external_data:
        # 检查配置中是否存在外部数据文件路径
        if "external_data_path" not in agent_conf:
            raise KeyError("配置中缺少 external_data_path 字段")
        # 将配置中的路径转换为绝对路径
        external_data_path = get_abs_path(agent_conf["external_data_path"])
        # 如果文件不存在，直接抛出异常
        if not os.path.exists(external_data_path):
            raise FileNotFoundError(f"外部数据文件不存在：{external_data_path}")
        # 以 utf-8 编码打开外部数据文件
        with open(external_data_path, "r", encoding="utf-8") as f:
            # 读取所有行，并跳过第一行表头
            for line in f.readlines()[1:]:
                # 去掉每行首尾空白后，按逗号分割字段
                arr = line.strip().split(",")

                # 提取各列字段，并去掉可能存在的双引号
                user_id = arr[0].replace('"', "")
                feature = arr[1].replace('"', "")
                efficiency = arr[2].replace('"', "")
                consumables = arr[3].replace('"', "")
                comparison = arr[4].replace('"', "")
                time = arr[5].replace('"', "")

                # 如果该用户在 external_data 中还不存在，则先初始化为空字典
                if user_id not in external_data:
                    external_data[user_id] = {}
                # 将当前用户在某个月份的使用记录保存到嵌套字典中
                external_data[user_id][time] = {
                    "特征": feature,
                    "效率": efficiency,
                    "耗材": consumables,
                    "对比": comparison,
                }
    # 返回已生成或已缓存的外部数据
    return external_data


# 将查询外部使用记录的函数注册为智能体工具
# 功能：根据用户ID和月份，获取对应的扫地/扫拖机器人使用记录
@tool(description="从外部系统中获取指定用户在指定月份的使用记录，未检索到时返回空字符串")
def fetch_external_data(user_id: str, month: str) -> str:
    # 先确保外部数据已经加载到内存中
    generate_external_data()
    try:
        # 返回指定用户在指定月份的使用记录
        return external_data[user_id][month]
    except KeyError:
        # 如果用户ID不存在，或者该用户在该月份没有记录，则记录警告日志
        logger.warn(f"[fetch_external_data]未能检索到用户:{user_id}在{month}的数据。")
        # 未命中数据时返回空字符串
        return ""

中间件和 Agent：

agent_tools：

@tool(description="调用后触发中间件自动为报告生成场景动态注入上下文信息，为后续提示词切换提供上下文支撑")
def fill_context_for_report():
    return "fill_context_for_report已调用"

middleware：

from typing import Callable
from langchain.agents import AgentState
from langchain.agents.middleware import wrap_tool_call, before_model, dynamic_prompt, ModelRequest
from langchain.tools.tool_node import ToolCallRequest
from langchain_core.messages import ToolMessage
from langgraph.runtime import Runtime
from langgraph.types import Command
from utils.logger_handler import logger
from utils.prompt_loader import load_system_prompts, load_report_prompts

# 工具调用拦截器
@wrap_tool_call
def monitor_tool(
        # request 表示当前这一次工具调用的请求对象, 里面包含工具名 参数 运行时上下文等信息
        request: ToolCallRequest,
        # handler 表示真正执行工具本体的函数
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
    # 记录当前执行的工具名称
    logger.info(f"[tool monitor]执行工具：{request.tool_call['name']}")
    # 记录当前工具接收到的参数
    logger.info(f"[tool monitor]传入参数：{request.tool_call['args']}")

    try:
        # 调用真正的工具执行逻辑
        result = handler(request)
        # 工具执行成功后记录日志
        logger.info(f"[tool monitor]工具{request.tool_call['name']}调用成功")
        # 如果当前工具是报告模式触发器, 就把 report 标记写入运行上下文, 后续 dynamic_prompt 会根据这个标记切换提示词
        if request.tool_call['name'] == "fill_context_for_report":
            request.runtime.context["report"] = True
        # 返回工具执行结果给 Agent 主流程继续使用
        return result

    except Exception as e:
        # 如果工具执行失败则记录异常原因
        logger.error(f"工具{request.tool_call['name']}调用失败，原因：{str(e)}")
        # 继续把异常抛出去 让上层知道这次工具调用失败了
        raise e


@before_model
def log_before_model(
        # state 表示当前 Agent 的运行状态
        state: AgentState,
        # request 这里实际接收的是运行时对象 Runtime, 可用于访问本轮运行的一些上下文信息
        request: Runtime,
):
    # 记录模型调用前当前一共积累了多少条消息, state["messages"] 即当前累计消息列表
    logger.info(f"[log_before_model]即将调用模型，带有{len(state['messages'])}条消息")

    # 打印最后一条消息的类型和内容
    logger.debug(f"[log_before_model]{type(state['messages'][-1]).__name__} | {state['messages'][-1].content.strip()}")

    return None


# 动态提示词选择器
@dynamic_prompt
def report_prompt_switch(request: ModelRequest):
    # 从本轮运行上下文中读取 report 标记, 如果没有这个字段 默认按 False 处理
    is_report = request.runtime.context.get("report", False)

    # 如果 report=True 说明当前已进入报告生成场景, 返回报告专用提示词
    if is_report:
        return load_report_prompts()

    # 否则返回默认系统提示词
    return load_system_prompts()

react_agent：

from langchain.agents import create_agent
from agent.tools.middleware import monitor_tool, log_before_model, report_prompt_switch
from agent.tools.agent_tools import (rag_summarize, get_weather, get_user_location, get_user_id, get_current_month,
                                     fetch_external_data, fill_context_for_report)
from model.factory import chat_model
from utils.prompt_loader import load_system_prompts

class ReactAgent(object):
    def __init__(self):
        # 创建一个带模型、工具和中间件的智能体
        self.agent = create_agent(
            model=chat_model,
            # 设置默认系统提示词
            system_prompt=load_system_prompts(),
            tools=[
                rag_summarize,
                get_weather,
                get_user_location,
                get_user_id,
                get_current_month,
                fetch_external_data,
                fill_context_for_report
            ],
            middleware=[monitor_tool, log_before_model, report_prompt_switch],
        )

    def execute_stream(self, query):
        # 把用户问题包装成消息格式
        input_dict = {
            "messages": [
                {"role": "user", "content": query},
            ]
        }

        # 以流式方式运行 Agent
        # input_dict：本轮输入消息
        # stream_mode="values"：按状态值流式返回执行结果
        # context={"report": False}：为本轮执行传入初始上下文
        # stream 不会一次性返回完整答案，而是会不断返回执行过程中的结果片段
        for chunk in self.agent.stream(input_dict, stream_mode="values", context={"report": False}):
            # 取出当前这一步中的最后一条消息
            latest_message = chunk["messages"][-1]

            # 如果这条消息有内容，就把它一段一段向外返回
            # yield 的作用是先返回这一段内容，但函数不结束，后面还能继续返回下一段
            if latest_message.content:
                yield latest_message.content.strip() + "\n"


if __name__ == "__main__":
    agent = ReactAgent()
    # execute_stream 因为用了 yield，所以返回的是一个生成器，不是普通字符串
    # 外层 for 会不断从生成器中取出一段段内容
    for chunk in agent.execute_stream("扫地机器人在我所在地区的气温下如何保养"):
        # 每取到一段内容就立刻打印，因此形成流式输出效果
        print(chunk, end="", flush=True)

交互层开发：

用户界面：

app：

import time

import streamlit as st
from agent.react_agent import ReactAgent


# 设置页面标题
st.title("智扫通机器人智能客服")
# 添加分隔线
st.divider()
# 初始化聊天记录
if "message" not in st.session_state:
    st.session_state["message"] = [
        {"role": "assistant", "content": "您好，我是智扫通机器人智能客服，请问有什么可以帮您？"}
    ]
# 初始化 Agent 实例
if "agent" not in st.session_state:
    st.session_state["agent"] = ReactAgent()
# 显示历史聊天消息
for message in st.session_state["message"]:
    st.chat_message(message["role"]).write(message["content"])
# 获取用户输入
prompt = st.chat_input()
# 如果用户提交了问题
if prompt:
    # 先显示用户消息
    st.chat_message("user").write(prompt)
    # 再保存用户消息到会话状态
    st.session_state["message"].append({"role": "user", "content": prompt})
    # 用于缓存助手返回的流式内容
    response_messages = []
    # 包装流式输出并实现打字机效果
    def capture(generator, cache_list):
        for chunk in generator:
            # 缓存每一段回答内容
            cache_list.append(chunk)
            # 按字符逐个输出
            for char in chunk:
                time.sleep(0.01)
                yield char
    # 显示等待提示
    with st.spinner("智能客服思考中..."):
        # 调用 Agent 的流式输出方法
        res_stream = st.session_state["agent"].execute_stream(prompt)
        # 实时显示助手回复
        st.chat_message("assistant").write_stream(capture(res_stream, response_messages))
    # 保存助手完整回复到聊天记录
    st.session_state["message"].append(
        {"role": "assistant", "content": response_messages[-1]}
    )
    # 刷新页面以更新界面状态
    st.rerun()