2026-04-08 手把手教你从零建立AI助手：RAG与Agent核心实战

import openai

response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "介绍一下LangChain框架的最新版本特性"}]
)
print(response.choices[0].message.content)

用户提问 → Thought（分析该做什么） → Action（调用工具） → Observation（获取结果） → 继续思考或给出最终答案

pip install langchain langchain-community langchain-openai chromadb faiss-cpu pypdf python-dotenv

 document_loader.py
import os
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_and_split(file_path: str):
    """加载PDF文档并切分成语义块"""
    loader = PyPDFLoader(file_path)   逐页加载，自动记录页码
    docs = loader.load()
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,         每块最大字符数
        chunk_overlap=100,       相邻块重叠，保留上下文连续性
    )
    chunks = splitter.split_documents(docs)
    return [chunk for chunk in chunks if chunk.page_content.strip()]

 vector_store.py
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

def build_vector_store(chunks):
    """将文档块转为向量并存储"""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = FAISS.from_documents(chunks, embeddings)
    return vector_store

 使用时
chunks = load_and_split("your_document.pdf")
vector_store = build_vector_store(chunks)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})   每次检索返回最相关的3个片段

 rag_chain.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser

 定义Prompt模板
prompt_template = ChatPromptTemplate.from_template("""
你是一个专业的技术助手。请基于以下参考资料回答用户的问题。
如果参考资料中没有相关信息，请如实告知，不要编造。

【参考资料】
{context}

【用户问题】
{question}

请给出准确、简洁的回答：
""")

 构建RAG链
llm = ChatOpenAI(model="gpt-4", temperature=0)   temperature=0保证答案稳定性

def format_docs(docs):
    return "\n\n".join([doc.page_content for doc in docs])

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt_template
    | llm
    | StrOutputParser()
)

 调用示例
answer = rag_chain.invoke("什么是LangChain？")
print(answer)

 agent_with_tools.py
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import tool

 定义工具1：天气查询
@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气（模拟API调用）"""
     实际生产环境中这里调用真实天气API
    weather_data = {"北京": "晴，25°C", "上海": "多云，22°C", "深圳": "阵雨，28°C"}
    return weather_data.get(city, f"未找到{city}的天气信息")

 定义工具2：计算器
@tool
def calculate(expression: str) -> str:
    """执行数学计算，输入如 '2 + 3  4'"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

 定义工具3：知识库检索（复用上面的RAG检索器）
@tool
def search_knowledge(query: str) -> str:
    """从知识库中检索相关文档内容"""
    docs = retriever.get_relevant_documents(query)
    if not docs:
        return "未找到相关信息"
    return "\n".join([doc.page_content[:500] for doc in docs[:2]])

 组装Agent
tools = [get_weather, calculate, search_knowledge]

agent = create_tool_calling_agent(
    llm=ChatOpenAI(model="gpt-4", temperature=0),
    tools=tools,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是一个智能助手，可以使用工具来回答用户问题。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

 测试：Agent自动规划执行
response = agent_executor.invoke({
    "input": "北京今天的天气怎么样？顺便帮我算一下 125  8 等于多少"
})
print(response["output"])