基于LangGraph+FastAPI + 混元大模型的企业智能助手RAG知识引擎系统设计与实践21.5

langgraph-enterprise-bot/
├── app/
│   ├── main.py                     # FastAPI入口：生命周期、路由注册、启动自检
│   ├── config.py                   # 全局配置中心：模型、向量库、缓存、检索参数
│   ├── api/                        # 接口路由层
│   │   ├── chat.py                 # 智能对话核心接口
│   │   ├── knowledge.py            # 知识库管理接口
│   │   ├── workflow.py             # 工作流状态管理接口
│   │   └── test.py                 # 服务自测接口
│   ├── core/                       # 核心能力引擎
│   │   ├── rag_engine.py           # RAG检索核心流水线
│   │   └── chat_memory.py          # Redis多轮对话记忆管理
│   ├── db/                         # 数据层封装
│   │   ├── vector_db.py            # 向量库统一操作封装
│   │   └── redis_db.py             # Redis异步连接池
│   ├── graph/                      # LangGraph工作流核心
│   │   ├── state.py                # 全局状态定义
│   │   ├── nodes.py                # 业务流程节点
│   │   ├── edges.py                # 条件分支路由
│   │   └── workflow_graph.py       # 工作流组装编译
│   ├── tools/                      # 企业业务工具集群
│   │   ├── rule_tool.py            # 公司制度查询工具
│   │   ├── ticket_tool.py          # 工单进度查询工具
│   │   └── user_tool.py            # 员工信息查询工具
│   ├── utils/                      # 通用工具函数
│   │   ├── logger.py               # 全链路日志配置
│   │   ├── doc_loader.py           # 多格式文档解析
│   │   └── validator.py            # 安全参数校验
│   └── static/
│       └── index.html              # 前端SPA可视化面板
├── tests/
│   └── test_service.py             # 全链路自动化测试脚本
├── data/                           # 企业知识库原始文档目录
│   ├── 员工手册.txt
│   ├── 产品FAQ.txt
│   └── IT支持指南.txt
├── docker-compose.yml              # 全服务容器编排配置
├── Dockerfile                      # 项目镜像构建文件
├── requirements.txt                # 依赖版本锁定清单
└── README.md                       # 项目部署使用文档

import os
from dotenv import load_dotenv

load_dotenv()

# ============================================================
# 腾讯混元大模型配置（TokenHub平台）
# ============================================================
LLM_API_KEY = os.getenv(
    "TENCENT_API_KEY",
    "sk-ymLiA.....",
)
LLM_BASE_URL = os.getenv(
    "LLM_BASE_URL",
    "https://tokenhub.tencentmaas.com/v1",
)
LLM_MODEL_NAME = os.getenv("LLM_MODEL_NAME", "hy3-preview")
LLM_TEMPERATURE = 0.1  # 低温度，保证业务回答严谨

# ============================================================
# 向量数据库配置（使用 Chroma 本地向量库）
# ============================================================
VECTOR_DB_TYPE = "chroma"
COLLECTION_NAME = "enterprise_knowledge"

# ============================================================
# 本地 Embedding 模型配置
# ============================================================
EMBEDDING_MODEL_PATH = os.getenv(
    "EMBEDDING_MODEL_PATH",
    r"D:\modelscope\hub\sentence-transformers\paraphrase-multilingual-MiniLM-L12-v2",
)

# ============================================================
# Redis 配置
# ============================================================
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
REDIS_DB = int(os.getenv("REDIS_DB", "0"))

# ============================================================
# 对话配置
# ============================================================
MAX_CHAT_HISTORY = 10
SIMILARITY_TOP_K = 5  # 检索 top-5 相似文档
SIMILARITY_THRESHOLD = 1.8  # L2 距离阈值，超过此值视为不相关（值越小越严格；归一化向量范围 0~2）

"""请求校验器 —— 输入安全校验 & 参数合法性检查。"""

import re
from typing import Optional

from pydantic import BaseModel, Field, field_validator

# ------------------------------------------------------------
# 校验规则常量
# ------------------------------------------------------------
MAX_QUERY_LENGTH = 2000
FORBIDDEN_PATTERNS = [
    re.compile(r"<script.*?>", re.IGNORECASE),
    re.compile(r"\{\{.*?\}\}"),
    re.compile(r"<%[^>]*%>"),
]


class ChatRequest(BaseModel):
    """对话请求模型（含校验）"""

    query: str = Field(..., min_length=1, max_length=MAX_QUERY_LENGTH, description="用户提问")
    chat_history: list = Field(default_factory=list, description="多轮对话历史")
    session_id: Optional[str] = Field(default=None, max_length=64, description="会话标识")

    @field_validator("query")
    @classmethod
    def sanitize_query(cls, v: str) -> str:
        """去除首尾空白，过滤潜在注入字符。"""
        v = v.strip()
        if not v:
            raise ValueError("query 不能为空")
        for pattern in FORBIDDEN_PATTERNS:
            v = pattern.sub("", v)
        return v

    @field_validator("session_id")
    @classmethod
    def validate_session_id(cls, v: Optional[str]) -> Optional[str]:
        if v and not re.fullmatch(r"[a-zA-Z0-9_-]+", v):
            raise ValueError("session_id 只能包含字母、数字、下划线和连字符")
        return v


class KnowledgeUploadRequest(BaseModel):
    """知识文档上传请求"""

    file_name: str = Field(..., min_length=1, max_length=255)
    content: str = Field(..., min_length=1)

    @field_validator("file_name")
    @classmethod
    def sanitize_filename(cls, v: str) -> str:
        # 只保留安全字符
        v = re.sub(r"[^\w.\-]", "_", v)
        if v.startswith("."):
            v = "_" + v
        return v

    @field_validator("content")
    @classmethod
    def validate_content(cls, v: str) -> str:
        if len(v) > 50_000:
            raise ValueError("单篇文档内容不能超过 50000 字符")
        return v

"""企业文档加载器 —— 支持 PDF / Markdown / TXT 等多种格式。"""

import os
from pathlib import Path
from typing import List

from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader
from langchain_core.documents import Document

from app.utils.logger import logger

DATA_DIR = Path(__file__).resolve().parent.parent.parent / "data"


def load_all_documents(data_dir: Path | None = None) -> List[Document]:
    """
    递归扫描 data/ 目录，加载所有支持格式的文档。

    支持格式：.pdf  .md  .txt
    """
    if data_dir is None:
        data_dir = DATA_DIR

    all_docs: List[Document] = []
    if not data_dir.exists():
        logger.warning("data/ 目录不存在，无文档可加载：%s", data_dir)
        return all_docs

    for file_path in data_dir.rglob("*"):
        if file_path.is_dir() or file_path.name.startswith("."):
            continue

        suffix = file_path.suffix.lower()
        try:
            if suffix == ".pdf":
                loader = PyPDFLoader(str(file_path))
            elif suffix == ".md":
                loader = UnstructuredMarkdownLoader(str(file_path))
            elif suffix == ".txt":
                loader = TextLoader(str(file_path), encoding="utf-8")
            else:
                continue

            docs = loader.load()
            for doc in docs:
                doc.metadata.setdefault("source", str(file_path))
            all_docs.extend(docs)
            logger.info("已加载文档：%s（%d 页/段）", file_path.name, len(docs))
        except Exception:
            logger.exception("文档加载失败：%s", file_path)

    logger.info("文档加载完成，共 %d 条记录", len(all_docs))
    return all_docs

"""对话记忆管理 —— 基于 Redis 的多轮对话历史存取。"""

import json
from typing import Dict, List, Optional

from app.config import MAX_CHAT_HISTORY
from app.db.redis_db import get_redis
from app.utils.logger import logger

KEY_PREFIX = "chat:history:"


async def save_message(session_id: str, role: str, content: str) -> None:
    """保存单条消息到会话历史。"""
    redis = await get_redis()
    key = f"{KEY_PREFIX}{session_id}"
    msg = json.dumps({"role": role, "content": content}, ensure_ascii=False)
    await redis.rpush(key, msg)
    await redis.ltrim(key, -MAX_CHAT_HISTORY * 2, -1)  # 保留最近N轮（每轮一问一答）
    await redis.expire(key, 3600 * 24)  # 24h 过期


async def get_history(session_id: str) -> List[Dict[str, str]]:
    """获取会话完整对话历史。"""
    redis = await get_redis()
    key = f"{KEY_PREFIX}{session_id}"
    raw = await redis.lrange(key, 0, -1)
    return [json.loads(m) for m in raw]


async def clear_history(session_id: str) -> None:
    """清除指定会话历史。"""
    redis = await get_redis()
    await redis.delete(f"{KEY_PREFIX}{session_id}")
    logger.info("已清除会话历史：%s", session_id)

"""企业规章制度查询工具 —— 检索公司内部条例与管理办法。"""

from langchain_core.tools import tool

from app.utils.logger import logger

# 企业规章制度库
_ENTERPRISE_RULES = {
    "请假": "员工请假需提前 1 个工作日在 OA 系统提交申请，3 天以内由直属领导审批，3 天以上需部门负责人加签。",
    "报销": "差旅报销需在行程结束后 7 个工作日内提交，附带正规发票和审批单，超标部分由个人承担。",
    "考勤": "弹性工作制：核心工作时间 10:00-17:00，月累计迟到超过 3 次将记入当月绩效。",
    "加班": "加班需提前报备，工作日加班按 1.5 倍时薪计算，休息日 2 倍，法定节假日 3 倍。",
    "信息安全": "禁止使用未经审批的外部存储设备；所有内部文档不得通过非授权渠道外传。",
}


@tool
def query_enterprise_rule(query: str) -> str:
    """
    企业规章制度与管理办法查询工具。

    Args:
        query: 制度关键词（如：请假、报销、考勤、加班、信息安全）

    Returns:
        匹配的制度条款原文或提示信息。
    """
    for keyword, rule_text in _ENTERPRISE_RULES.items():
        if keyword in query:
            logger.info("命中企业制度：%s", keyword)
            return f"【{keyword}】{rule_text}"
    logger.info("未命中企业制度，query=%s", query)
    return "未找到相关制度条款，建议查阅《公司员工手册》或咨询 HR 部门。"

from langchain_core.tools import tool
import requests

from app.utils.logger import logger


@tool
def query_enterprise_ticket(ticket_id: str) -> str:
    """
    企业工单状态查询工具。

    Args:
        ticket_id: 工单编号

    Returns:
        工单详细状态、处理进度、处理人。
    """
    try:
        # 对接企业内部工单系统 API（真实业务接口）
        res = requests.get(
            f"http://enterprise-api.local/ticket/{ticket_id}", timeout=5
        )
        if res.status_code == 200:
            data = res.json()
            return (
                f"工单{ticket_id}状态：{data['status']}，"
                f"处理人：{data['operator']}，"
                f"进度：{data['progress']}"
            )
        logger.warning("工单查询失败：%s, status=%d", ticket_id, res.status_code)
        return f"工单{ticket_id}查询失败，暂无数据"
    except Exception as e:
        logger.exception("工单查询异常：%s", ticket_id)
        return f"工单查询异常：{str(e)}，请人工核查"

"""用户信息查询工具 —— 对接企业 HR / 通讯录系统。"""

from langchain_core.tools import tool

from app.utils.logger import logger

# 模拟的用户信息库（生产环境对接企业 LDAP / HR 系统）
_MOCK_USERS = {
    "zhangsan": {"name": "张三", "department": "技术部", "position": "高级工程师", "email": "zhangsan@company.com"},
    "lisi": {"name": "李四", "department": "产品部", "position": "产品经理", "email": "lisi@company.com"},
    "wangwu": {"name": "王五", "department": "运营部", "position": "运营总监", "email": "wangwu@company.com"},
}


@tool
def query_user_info(user_identifier: str) -> str:
    """
    企业用户信息查询工具。

    Args:
        user_identifier: 员工工号 / 用户名 / 邮箱前缀

    Returns:
        用户部门、职位、联系方式等信息。
    """
    user = _MOCK_USERS.get(user_identifier.lower())
    if user:
        logger.info("用户查询成功：%s", user_identifier)
        return (
            f"姓名：{user['name']} | "
            f"部门：{user['department']} | "
            f"职位：{user['position']} | "
            f"邮箱：{user['email']}"
        )
    logger.info("未找到用户：%s", user_identifier)
    return f"未查询到 {user_identifier} 的员工信息，请核对账号后重试。"

from typing import Any, Dict, List, Optional, TypedDict


class GraphState(TypedDict, total=False):
    """企业工作流全局状态 —— TypedDict 与 LangGraph StateGraph 原生兼容。"""

    user_query: str
    chat_history: List[Dict[str, str]]
    intent_type: Optional[str]
    retrieve_docs: List[Any]
    tool_result: Optional[str]
    answer: Optional[str]
    error_msg: Optional[str]
    need_human: bool


def create_initial_state(
    user_query: str = "",
    chat_history: Optional[List[Dict[str, str]]] = None,
) -> GraphState:
    """创建初始状态字典（节点入口可用）。"""
    return GraphState(
        user_query=user_query,
        chat_history=chat_history or [],
        intent_type=None,
        retrieve_docs=[],
        tool_result=None,
        answer=None,
        error_msg=None,
        need_human=False,
    )

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

from app.config import LLM_API_KEY, LLM_BASE_URL, LLM_MODEL_NAME, LLM_TEMPERATURE
from app.core.rag_engine import retrieve_knowledge_docs
from app.graph.state import GraphState
from app.tools.ticket_tool import query_enterprise_ticket

# ---------- 初始化大模型 ----------
llm = ChatOpenAI(
    api_key=LLM_API_KEY,
    base_url=LLM_BASE_URL,
    model=LLM_MODEL_NAME,
    temperature=LLM_TEMPERATURE,
)


# =============================================================
# 1. 意图识别节点
# =============================================================
def intent_recognize_node(state: GraphState) -> GraphState:
    """识别用户提问意图：知识问答 / 业务工具 / 未知问题。"""
    prompt = ChatPromptTemplate.from_template("""\
你是企业智能助手，请识别用户提问意图，仅返回：knowledge / tool / unknown

规则：
1. 询问公司制度、手册、流程规范、产品知识 → knowledge
2. 查询工单、数据、业务办理 → tool
3. 无关问题、模糊问题 → unknown

用户提问：{user_query}
""")
    chain = prompt | llm
    intent = chain.invoke({"user_query": state["user_query"]}).content.strip()
    new_state = dict(state)
    new_state["intent_type"] = intent
    return new_state


# =============================================================
# 2. 知识检索节点
# =============================================================
def knowledge_retrieve_node(state: GraphState) -> GraphState:
    """RAG 知识库检索。"""
    docs = retrieve_knowledge_docs(state["user_query"])
    new_state = dict(state)
    new_state["retrieve_docs"] = docs
    return new_state


# =============================================================
# 3. 工具调用节点
# =============================================================
def tool_invoke_node(state: GraphState) -> GraphState:
    """根据提问调用对应业务工具。"""
    import re

    new_state = dict(state)
    try:
        if "工单" in state["user_query"]:
            ticket_ids = re.findall(r"T\d+", state["user_query"])
            if ticket_ids:
                res = query_enterprise_ticket.invoke(ticket_ids[0])
                new_state["tool_result"] = res
                return new_state
        new_state["tool_result"] = "未识别到有效业务参数，无法查询"
    except Exception as e:
        new_state["error_msg"] = str(e)
        new_state["need_human"] = True
    return new_state


# =============================================================
# 4. 答案生成节点
# =============================================================
def generate_answer_node(state: GraphState) -> GraphState:
    """结合检索结果 / 工具结果生成最终答案。"""
    context = ""
    if state.get("retrieve_docs"):
        context = "\n".join([doc.page_content for doc in state["retrieve_docs"]])
    if state.get("tool_result"):
        context += f"\n业务查询结果：{state['tool_result']}"

    prompt = ChatPromptTemplate.from_template("""\
你是专业的企业内部智能助手，基于以下参考内容回答用户问题，回答严谨、简洁、准确，禁止编造信息。

参考内容：{context}

用户问题：{user_query}
对话历史：{chat_history}
""")
    chain = prompt | llm
    answer = chain.invoke({
        "context": context,
        "user_query": state["user_query"],
        "chat_history": state.get("chat_history", []),
    }).content

    new_state = dict(state)
    new_state["answer"] = answer
    return new_state


# =============================================================
# 5. 人工兜底节点
# =============================================================
def human_fallback_node(state: GraphState) -> GraphState:
    """异常场景人工介入兜底。"""
    new_state = dict(state)
    new_state["answer"] = (
        f"当前问题暂时无法自动处理，已为您转接人工客服，"
        f"异常信息：{state.get('error_msg', '未知异常')}"
    )
    return new_state

from app.graph.state import GraphState


def intent_route_edge(state: GraphState) -> str:
    """意图分支路由。"""
    intent = state.get("intent_type", "")
    if intent == "knowledge":
        return "knowledge_retrieve"
    elif intent == "tool":
        return "tool_invoke"
    return "generate_answer"


def error_route_edge(state: GraphState) -> str:
    """异常分支路由。"""
    if state.get("need_human") or state.get("error_msg"):
        return "human_fallback"
    return "end"

from langgraph.graph import StateGraph, END

from app.graph.edges import error_route_edge, intent_route_edge
from app.graph.nodes import (
    generate_answer_node,
    human_fallback_node,
    intent_recognize_node,
    knowledge_retrieve_node,
    tool_invoke_node,
)
from app.graph.state import GraphState


def build_workflow_graph():
    """构建并编译企业级 LangGraph 工作流。"""
    # 初始化状态图
    graph = StateGraph(GraphState)

    # 注册所有节点
    graph.add_node("intent_recognize", intent_recognize_node)
    graph.add_node("knowledge_retrieve", knowledge_retrieve_node)
    graph.add_node("tool_invoke", tool_invoke_node)
    graph.add_node("generate_answer", generate_answer_node)
    graph.add_node("human_fallback", human_fallback_node)

    # 设置入口节点
    graph.set_entry_point("intent_recognize")

    # 配置意图分支流转
    graph.add_conditional_edges(
        "intent_recognize",
        intent_route_edge,
        {
            "knowledge_retrieve": "knowledge_retrieve",
            "tool_invoke": "tool_invoke",
            "generate_answer": "generate_answer",
        },
    )

    # 固定节点流转：检索/工具 → 答案生成
    graph.add_edge("knowledge_retrieve", "generate_answer")
    graph.add_edge("tool_invoke", "generate_answer")

    # 异常兜底分支
    graph.add_conditional_edges(
        "generate_answer",
        error_route_edge,
        {
            "human_fallback": "human_fallback",
            "end": END,
        },
    )

    graph.add_edge("human_fallback", END)

    # 编译图
    return graph.compile()


# 全局单例工作流
enterprise_workflow = build_workflow_graph()

# 1. 安装项目依赖
pip install -r requirements.txt

# 2. 配置API密钥（环境变量）
export API_KEY="你的API密钥"

# 3. 启动本地Redis缓存
docker run -d --name redis -p 6379:6379 redis:7-alpine

# 4. 启动后端服务
python -m uvicorn app.main:app --host 0.0.0.0 --port 8001

# 5. 访问前端页面
# 浏览器打开：http://localhost:8001

# 一键启动所有服务
docker compose up -d

# 查看服务运行日志
docker compose logs -f app

# 停止所有服务
docker compose down

# 全量测试
python tests/test_service.py

# 指定服务地址测试
python tests/test_service.py --url http://localhost:8001

# 跳过耗时测试项
python tests/test_service.py --skip-heavy

# 输出详细测试日志
python tests/test_service.py -v