实用指南：利用RAG、Agent和LangGraph构建企业级AI助手
《实用笔记》操作指南：使用RAG、Agents和LangGraph构建企业级AI助手——为采用该架构的团队提供的合同模板、检查清单以及可直接插入的代码片段。
来源: Building a Company AI Assistant with RAG, Agents, and LangGraph
以下笔记为“利用RAG、Agents和LangGraph构建企业级AI助手”提供了实用的操作路径。重点在于合同定义、检查项以及可直接插入的代码占位符，而非激励性表述。 在阅读概述部分时，首先列出合同要求：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持清晰。 建议采用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一责任模块，而非复杂的流程链。
rag-agent-project/
│
├── documents/
├── faiss_index/
│
├── .env
├── .gitignore
├── requirements.txt
│
├── config.py
├── ingest.py
├── rag.py
├── query_rewriter.py
├── reranker.py
├── tools.py
├── graph.py
├── main.py
└── app.py
pip install -r requirements.txt
.env
OPENAI_API_KEY=your_openai_api_key
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
raise ValueError("OPEN_API_KEY is not set."
"Add it to yor .env file.")
LLM_MODEL = "gpt-4.1-mini"
EMBEDDING_MODEL = "text-embedding-3-small"
HUGGINGFACE_EMBEDDING = "sentence-transformers/all-MiniLM-L6-v2"
FAISS_PATH = "faiss_index"
DATABASE_PATH = "company_assistant.db"
documents/
├── employee_handbook.pdf
├── hr_policy.pdf
└── it_policy.pdf
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from config import HUGGINGFACE_EMBEDDING , FAISS_PATH
DOCUMENTS_DIR = Path("documents")
def load_documents():
all_documents = []
pdf_files = list(DOCUMENTS_DIR.glob("*.pdf"))
if not pdf_files:
raise FileNotFoundError(
"No pdf file found in documents/"
)
for pdf_file in pdf_files:
print(f"Loading: {pdf_file}")
loader = PyPDFLoader(str(pdf_file))
documents = loader.load()
for document in documents:
document.metadata["source"] = pdf_file.name
if "handbook" in pdf_file.name.lower():
document.metadata["document_type"] = "handbook"
elif "policy" in pdf_file.name.lower():
document.metadata["document_type"] = "policy"
else:
document.metadata["document_type"] = "other"
all_documents.append(document)
return all_documents
def main():
# 1. Load PDFs
documents = load_documents()
print(f"\nLoaded {len(documents)} pages")
# 2. Split documents
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100
)
chunks = text_splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks.")
# 3. Create embeddings
embeddings = HuggingFaceEmbeddings(
model_name=HUGGINGFACE_EMBEDDING
)
# 4. Create FAISS database
vector_store = FAISS.from_documents(
chunks,
embeddings
)
# 5. Save FAISS database
vector_store.save_local(FAISS_PATH)
print(
f"\nFAISS index saved to: {FAISS_PATH}"
)
if __name__ == "__main__":
main()
Loading: documents/employee_handbook.pdf
Loading: documents/hr_policy.pdf
Loading: documents/it_policy.pdf
Loaded 20 pages.
Created 60 chunks.
FAISS index created successfully!
Chunk 1
-------------------------
A B C D E F G H I J
|
| overlap
v
Chunk 2
E F G H I J K L M N
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from config import(
HUGGINGFACE_EMBEDDING,
FAISS_PATH
)
embeddings = HuggingFaceEmbeddings(
model_name=HUGGINGFACE_EMBEDDING
)
vector_store = FAISS.load_local(
FAISS_PATH,
embeddings,
allow_dangerous_deserialization=True
)
from rag import vector_store
from langchain_huggingface import (HuggingFaceEmbeddings,HuggingFacePipeline)
from transformers import pipeline
pipe = pipeline(
"text-generation",
model="Qwen/Qwen2.5-0.5B-Instruct",
max_new_tokens=256,
temperature=0.1
)
llm = HuggingFacePipeline(
pipeline=pipe
)
def rewrite_query(question:str)-> str:
prompt = f"""
Rewrite the user's question into a concise search query for company document database.
The query should contain important keywords that are likely to appear in company documents.
Do not answer the question.
User question:{question}
search query:
"""
response = llm.invoke(prompt)
return response
Chunk 1
Chunk 2
Chunk 3
Chunk 4
Chunk 5
from langchain_huggingface import (HuggingFaceEmbeddings,HuggingFacePipeline)
from transformers import pipeline
pipe = pipeline(
"text-generation",
model="Qwen/Qwen2.5-0.5B-Instruct",
max_new_tokens=256,
temperature=0.1
)
llm = HuggingFacePipeline(
pipeline=pipe
)
def rerank_documents(query:str, documents:list,top_n: int = 3):
scored_documents =[]
for document in documents:
prompt = f"""
You are evaluating document releveance.
Rate how relevant the document is to the user's question.
Return ONLY an intefer from 0 to 10.
0 = competely irrelevent
10 = extreamly relevent
Question: {query}
Document:{document.page_content}
"""
response = llm.invoke(prompt)
try:
score = int(response.content.strip())
except ValueError:
score = 0
scored_documents.append((score,document))
#Highest score first
scored_documents.sort(
key=lambda item:item[0],
reverse=True
)
return[
document
for score, document
in scored_documents[:top_n]
]
from langchain.tools import tool
from rag import vector_store
from query_rewriter import(rewrite_query)
from reranker import (rerank_documents)
@tool
def company_document_search(query: str) -> str:
"""
Search company PDF documents for information about
company policies, employee benifits, leave, holidays, work from home,
IT rules, working hours, and othher internal information.
"""
#1. Rewrite Query
imporved_query = rewrite_query(query)
print(f"\n[Rewritten Query]"
f"\n{imporoved_query}")
#2. Vector serach
documents = (vector_store.similarity_search(
imporved_query,
k=5
))
if not documents:
return(
"No relevent information"
"was found in the company documents"
)
#Reranking
documents = rerank_documents(
imporved_query, documents,top_n=3
)
#4.Formate context
result = []
for document in documents:
source = document.metadata.get("source","Unknown")
page = document.metadata.get("page","Unknown")
#PyPDFLoader user zero-based pages
if isinstance(page, int):
page = page+1
results.append(f"""
SOURCE:{source}
PAGE:{page}
CONTENT: {document.page_content}
""")
return "\n".join(result)
@tool
def calculator(expression: str) -> str:
"""
Calculate a mathematical expression.
"""
try:
# result = eval(expression)
# for learning only
# Do not use eval() in production
allowed = ("0123456789""+-*/(). ")
if not all(char in allowed for char in expression):
return ("Invalid mathmatical expression")
result = eval(expression,
{"__builtins__":{}},
{}
)
return str(result)
except Exception:
return "Unable to calculate the expression."
from langchain_ollama import ChatOllama
from langgraph.graph import (
StateGraph,
MessagesState,
START
)
from langgraph.prebuilt import (
ToolNode,
tools_condition
)
from langgraph.checkpoint.memory import InMemorySaver
from tools import (
company_document_search,
calculator
)
# Local Qwen model
llm = ChatOllama(
model="qwen2.5:3b",
temperature=0
)
# Tools
tools = [
company_document_search,
calculator
]
llm_with_tools = llm.bind_tools(tools)
# Agent
def agent_node(state: MessagesState):
response = llm_with_tools.invoke(
state["messages"]
)
return {
"messages": [response]
}
# Graph
graph_builder = StateGraph(MessagesState)
graph_builder.add_node(
"agent",
agent_node
)
graph_builder.add_node(
"tools",
ToolNode(tools)
)
# START → Agent
graph_builder.add_edge(
START,
"agent"
)
# Agent → Tool OR END
graph_builder.add_conditional_edges(
"agent",
tools_condition
)
# Tool → Agent
graph_builder.add_edge(
"tools",
"agent"
)
# Checkpointer
checkpointer = InMemorySaver()
# Compile
graph = graph_builder.compile(
checkpointer=checkpointer
)
SOURCE: hr_policy.pdf
PAGE: 4
Source: hr_policy.pdf, page 4.
from langchain_core.messages import HumanMessage, SystemMessage
from graph import graph
SYSTEM_PROMPT="""
You are an intelligent company assistant.
You answer question using company documents and avalable tools.
Rules:
1. Use company_document_search whenever the user asks about company-specific information.
2. Never invent company policies, benifits, rules, or procedures.
3. Use the calculator when mathmatical calculation are required.
4. When company documents are used, mention the source document and page.
5. If the company documents do not contain the requested information, clearly say
that you could not find it.
6. Use previous conversation context to undersatnd follow-up questions.
7. Give concise and clear answer
"""
def chat(
question:str,
thread_id: str):
config = {
"configurable":{"thread_id":thread_id}
}
#Check whether thhis conversation already exists
state = graph.get_state(config)
if not state.values:
messages = [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=question)
]
else:
messages=[
HumanMessage(content=question)]
result = graph.invoke({
"messages":messages},
config=config
)
return result["messages"][-1].content
def main():
print("===========================================")
print("CoOMPANY AI ASSISTANT")
print("===========================================")
print("\nType 'exist' to quit.")
thread_id = input("\nEnter conversation ID:")
while True:
question = input("\nYou: ")
if question.lower() in [
"exit",
"quit"
]:
break
try:
answer = chat(question,thread_id)
print(f"\nAI: {answer}")
except Exception as e:
print(f"\nError: {e}")
if __name__ == "__main__":
main()
import streamlit as st
from langchain_core.messages import (
HumanMessage,
AIMessage,
SystemMessage
)
from graph import graph
# Page configuration
st.set_page_config(
page_title="Company AI Assistant",
page_icon="🤖",
layout="centered"
)
# System prompt
SYSTEM_PROMPT = """
You are an intelligent company assistant.
You answer questions using company documents
and available tools.
Rules:
1. Use company_document_search whenever
the user asks about company-specific
information.
2. Never invent company policies,
benefits, rules, or procedures.
3. Use the calculator when mathematical
calculations are required.
4. When company documents are used,
mention the source document and page.
5. If the company documents do not contain
the requested information, clearly say
that you could not find it.
6. Use previous conversation context to
understand follow-up questions.
7. Give concise and clear answers.
"""
# Sidebar
st.sidebar.title("⚙️ Settings")
thread_id = st.sidebar.text_input(
"Conversation ID",
value="streamlit-user"
)
st.sidebar.markdown(
"""
### About
This application demonstrates:
- RAG
- AI Agents
- LangGraph
- FAISS
- Query rewriting
- Reranking
- Conversation memory
- Tool calling
"""
)
# Main UI
st.title("🏢 Company AI Assistant")
st.write(
"Ask questions about company policies, "
"benefits, leave, IT rules, and more."
)
# Chat history in Streamlit
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
# Display previous messages
for message in st.session_state.chat_history:
with st.chat_message(
message["role"]
):
st.markdown(
message["content"]
)
# User input
question = st.chat_input(
"Ask something about the company..."
)
if question:
# Display user question
with st.chat_message("user"):
st.markdown(question)
st.session_state.chat_history.append(
{
"role": "user",
"content": question
}
)
# Build messages
config = {
"configurable": {
"thread_id": thread_id
}
}
# Check whether this thread
# already has state
state = graph.get_state(
config
)
if not state.values:
messages = [
SystemMessage(
content=SYSTEM_PROMPT
),
HumanMessage(
content=question
)
]
else:
messages = [
HumanMessage(
content=question
)
]
# Call LangGraph
with st.chat_message("assistant"):
with st.spinner(
"Thinking..."
):
try:
result = graph.invoke(
{
"messages": messages
},
config=config
)
answer = result[
"messages"
][-1].content
st.markdown(answer)
# Save answer
st.session_state.chat_history.append(
{
"role": "assistant",
"content": answer
}
)
except Exception as e:
st.error(
f"Something went wrong: {e}"
)
操作检查清单
在修改代码之前，需为操作检查清单明确输入项、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新执行相应步骤，而无需猜测隐藏状态。
需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的一部分，而非后续补充的内容。
需引用实际作为答案依据的段落。若没有引用，操作人员就无法区分是虚假信息还是索引缺失导致的错误。
在成本较高的操作之后设置检查点。当操作员重新尝试后续节点时，续订服务不应再次对同一次大型语言模型调用收费。
锁定依赖版本，并记录用于运行演示的镜像摘要。可重复性比经验知识更为可靠。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。
在推广该技术栈之前，先冻结版本，为关键路径保存标准记录，并确认回滚步骤。共享环境需要设置速率限制、租户检查机制，以及明确的密钥轮换负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。
针对656e1cd26886的批量说明：将提供商密钥存放在仓库之外，为每个会话设置令牌使用上限，并将记录与评估相关文件放在一起，以便后续更换模型时仍能保持对比性。
针对强化措施第0条，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。
强化措施细节0/780：需统计该措施的运行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非主观判断来决定是否保留该变更。
在处理强化措施笔记1时，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 相比冗长的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。
强化措施细节1/780：针对该笔记，需测量执行时间、错误类型以及代币消耗情况，然后依据固定的评估标准而非主观感受来决定是否保留该修改。
将强化措施笔记2视为可度量的对象会更有助于其有效实施。在扩大范围之前，先记录一份理想的运行示例、一个失败案例以及回滚说明。 在功能结果旁同时记录执行时间以及代币或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外支出。
强化措施细节2/780：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
对于强化措施记录3，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。同时需将正常流程与故障恢复流程一并记录下来。重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。
强化措施细节3/780：为该记录测量运行时间、错误类型以及令牌消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在处理强化措施第4条时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合约定。 将此阶段视为输入与验证后输出之间的契约。为相关产物命名，明确成功判定标准，杜绝默许的部分完成情况。
强化措施细节4/780：需测量该条措施的执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该修改。
将强化措施第5条视为可量化的目标来处理效果最佳。在扩大范围之前，先记录一份理想的运行日志、一个失败案例以及回滚说明。 配置信息应与应用程序代码分开存放。环境文件、密钥存储和功能开关应集中于一个位置，以便操作人员无需查看全部代码结构即可进行审计。
强化措施细节5/780：为该记录测量运行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
对于强化措施记录6，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比复杂的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能指向单一责任方，而非混乱的整个流程。
强化措施细节6/780：为该记录测量运行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
在处理强化措施笔记7时，首先写下相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在代码从演示环境转向共享环境时出现意外费用。
强化措施细节7/780：为该笔记测量实际执行时间、错误类型以及令牌消耗情况，然后依据固定的评估标准而非个人经验来决定是否保留该修改。
将强化措施笔记8视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一个理想运行案例、一个失败案例以及回滚说明。 同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
强化措施细节 8/780：记录该步骤的运行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。
对于强化措施笔记 9，在修改代码之前需明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应将此阶段视为输入参数与验证后输出结果之间的契约，为相关成果命名、定义成功检测标准，并拒绝默许的半完成状态。
强化措施细节 9/780：记录该步骤的运行时间、错误类型以及代币消耗情况，然后依据固定的问题集而非个人经验来判断是否保留该变更。