LangChain驱动的智能体AI 第一部分

本指南将逐步构建从原材料到可运行系统的完整流程，内容为：使用 LangChain 的智能体 AI —— 第一部分：构建语义 RAG 流水线。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在修改代码之前，应先明确输入内容、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。

技术栈

在规划技术栈时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 相比冗长的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障点应指向单一职责，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。

1. 系统配置

在处理“1. 系统配置”时，首先写下相关契约：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. 项目结构

在处理“2. 项目结构”时，首先需写下接口规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本有助于避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── chains.py           # llm chains
├── nodes.py            # python functions utlizing llm chains
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

在处理“2. 项目结构”时，首先需写下接口规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理措施都是产品的一部分，而非后续需要补充的功能。

2.1. LLM链

2.1. 将大语言模型链视为可度量的系统时，其效果最佳。在扩大应用范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。相较于复杂的脚本，应优先选择小型且易于测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。需为每轮对话和每次会话设定Token预算——智能工具往往会大量消耗上下文，设置上限可避免演示过程变成意外的费用账单。

2.2. 语义RAG

2.2. 将语义RAG视为可度量的界面时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，杜绝默默完成部分任务的情况。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。

2.3. FastAPI后端

2.3. 将 FastAPI 后端视为可度量的系统时，其性能表现最佳。在扩展功能范围之前，需记录一份理想的操作流程、一个故障案例以及回滚说明。在功能结果之外，还需记录处理时间以及令牌或查询成本。提前了解这些成本信息，可避免在系统从演示环境过渡到共享环境时出现意外费用。应将分块策略与检索策略分开处理，当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。

2.4. Streamlit 客户端

对于2.4版本：在修改代码之前，需先定义输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体责任方，而非整个复杂的流程。必须引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。

3. 使用LCEL的LLM链

对于使用 LCEL 的第 3 类 LLM 链，应在修改代码之前明确输入内容、各步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 当下一步操作为编写代码或调用工具时，优先采用具有模式验证的结构化输出，而非自由形式的文字描述。

3.1. LangChain 表达语言（LCEL）

对于3.1版本的LangChain表达语言（LCEL），在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果之外还需记录执行时间以及token或查询成本。提前显示成本可避免在流程从演示环境切换到共享环境时出现意外费用。 需注明实际用于生成答案的对应内容。如果没有引用依据，操作员就无法区分是幻觉内容还是索引缺失导致的错误。 对于3.1版本的LangChain表达语言（LCEL），在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品功能的一部分，而非事后才处理的措施。

r抛光。

class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""

    binary_score: bool = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


async def retrieval_grader(document: Document, question: str):
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    structured_llm_grader = llm.with_structured_output(GradeDocuments)

    system = """You are a grader assessing relevance of a retrieved document to a user question. \n
    If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.\n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
    if the document is relevant, give it a score 'yes', otherwise give it a score 'no'
    """

    grade_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system),
            ("human", "Retrieved document: \n\n {document} User question: {question}"),

        ]
    )

    retrieval_grader = grade_prompt | structured_llm_grader
    grade = await retrieval_grader.ainvoke({"document": document, "question": question})

    return grade.binary_score

4. 使用PineCone实现语义RAG

4.1. 数据摄取流程

4.2>检索流程

4.3. VectorService类

4.3.1. 初始化

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore

class VectorService:
    def __init__(self, index_name: str = "knowledgebase"):
        self.index_name = index_name

        # Embedding model
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key=api_key)

        # Pinecone vectorstore wrapper
        self.vectorstore = PineconeVectorStore(
            index_name=index_name,
            embedding=self.embeddings,
            pinecone_api_key=pinecone_key,
        )

        # Retriever
        self.retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3}
        )

        # Text splitter (LangChain-native)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=["\n\n", "\n", ".", " ", ""]
        )

4.3.2.存储文件内容

    async def store_file_content_in_db(self, filepath: str) -> None:
        """
        Read the uploaded text file, split the file, embed chunks and save chunks to vector database
        :param filepath: filepath to load to vectordb
        :return: None
        """
        logger.debug(f"Loading file: {filepath}")

        # Load entire file (safe for most RAG use cases)
        async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
            raw_text = await f.read()

        # Minimal cleaning (preserve structure)
        raw_text = raw_text.replace("\r", "")

        # Split into LangChain Documents
        docs = self.splitter.create_documents(
            [raw_text],
            metadatas=[{"source": os.path.basename(filepath)}]
        )

        logger.debug(f"Storing {len(docs)} chunks into Pinecone")

        # Ingest into Pinecone
        PineconeVectorStore.from_documents(
            documents=docs,
            embedding=self.embeddings,
            index_name=self.index_name,
            pinecone_api_key=pinecone_key,
        )

4.3.3.搜索文档

    async def search_documents(self, query: str) -> list[Document]:
        logger.debug(f"Searching for: {query}")
        return self.retriever.invoke(query)

5. FastAPI后端

5.1.文件上传功能

async def save_file(file: UploadFile) -> str:
    """
    accept a UploadFile from FastAPI and save the file in specified location
    :param file: UploadFile specified by users from fastAPI interface
    :return: path of the saved file
    """
    await makedirs("uploads", exist_ok=True)
    filepath = os.path.join("uploads", file.filename)
    async with aiofiles.open(filepath, "wb") as f:
        while chunk := await file.read(DEFAULT_CHUNK_SIZE):
            await f.write(chunk)
    return filepath

5.2.请求/响应模型

class RAGRequest(BaseModel):
    prompt: str

class RAGResponse(BaseModel):
    answer: str
    num_original_documents: int

5.3>答案生成接口

# dependencies.py
from fastapi import Body, HTTPException
from schemas import RAGRequest
from nodes import generate

async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await generate(body.prompt)

        return {
            "answer": generation.get("generation", ""),
            "num_original_documents": generation.get("num_original_documents", 0)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# main.py
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(generation: dict = Depends(get_generation)) -> RAGResponse:
    return RAGResponse(**generation)

5.4>后台处理

@app.post("/upload")
async def file_upload_controller(
        file: Annotated[UploadFile, File(description="Uploaded PDF documents")],
        bg_text_processor: BackgroundTasks,
):
    if file.content_type != "application/pdf":
        raise HTTPException(
            detail=f"Only uploading PDF documents are supported",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
    try:
        filepath = await save_file(file)
        bg_text_processor.add_task(pdf_text_extractor, filepath)
        bg_text_processor.add_task(
            vector_service.store_file_content_in_db,
            filepath.replace("pdf", "txt")
        )
    except Exception as e:
        raise HTTPException(
            detail=f"An error occurred while saving file - Error: {e}",
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    return {"filename": file.filename, "message": "File uploaded successfully"}

6. Streamlit前端

6.1>文件上传功能

st.write("Upload a file to FastAPI")
file = st.file_uploader("Choose a file", type=["pdf"])

if st.button("Submit"):
    if file is not None:
        files = {"file": (file.name, file, file.type)}
        response = requests.post("http://localhost:8000/upload", files=files)
        st.write(response.text)
    else:
        st.write("No file uploaded.")

6.2>对话历史记录

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

6.3>聊天流程

if prompt := st.chat_input("Write your prompt in this input field"):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.text(prompt)

    question = prompt

    response = requests.post(
        f"http://localhost:8000/generate_text",
        json={"prompt": question}
    )
    response.raise_for_status()

    response_json = response.json()
    answer = response_json["answer"]
    st.session_state.messages.append({"role": "assistant", "content": answer})

    with st.chat_message("assistant"):
        st.markdown(answer)

7.总结生成

操作检查清单