构建 Agentic RAG 应用

pip install -U langchain-google-genai langchain-voyageai langchain-core langchain-community

import os


# MyScale API凭据
os.environ["MYSCALE_HOST"] = "msc-24862074.us-east-1.aws.myscale.com"
os.environ["MYSCALE_PORT"] = "443"
os.environ["MYSCALE_USERNAME"] = "your_myscale_username"
os.environ["MYSCALE_PASSWORD"] = "your_myscale_password"


# Tavily API密钥
os.environ["TAVILY_API_KEY"] = "your_tavily_api_key"


# Gemini的Google API密钥
os.environ["GOOGLE_API_KEY"] = "your_google_api_key"

from langchain_text_splitters import CharacterTextSplitter


with open("myscaledb_summary.txt") as f:
    state_of_the_union = f.read()


text_splitter = CharacterTextSplitter(
    separator="\\n\\n",
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    is_separator_regex=False,
)


texts = text_splitter.create_documents([state_of_the_union])

from langchain_voyageai import VoyageAIEmbeddings
from langchain_community.vectorstores import MyScale


embeddings = VoyageAIEmbeddings(
    voyage_api_key="your_voyageai_api_key",
    model="voyage-law-2"
)


vectorstore = MyScale.from_documents(
    texts,
    embeddings,
)


retriever = vectorstore.as_retriever()

from langchain.tools.retriever import create_retriever_tool


retriever_tool = create_retriever_tool(
    retriever,
    "retrieve_myscale_content",
    "用于返回有关MyScaleDB的信息。",
)

from langchain_community.tools import TavilySearchResults


tool = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
    include_answer=True,
    name="live_search",
    description="用于搜索互联网上的最新新闻。",
)


tools = [retriever_tool, tool]

def grade_documents(state) -> Literal["generate", "rewrite"]:
    # 评分的数据模型
    class Grade(BaseModel):
        """用于相关性检查的二进制分数。"""
        binary_score: str = Field(description="相关性分数 'yes' 或 'no'")


    # LLM
    model = ChatGoogleGenerativeAI(
        model="gemini-1.5-pro",
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )
    
    llm_with_tool = model.with_structured_output(Grade)


    # 提示
    prompt = PromptTemplate(
        template="""你是一个评分员，正在评估检索到的文档与用户问题的相关性。
        这是检索到的文档：


        {context}


        这是用户问题：{question}
        如果文档包含与用户问题相关的关键词或语义意义，请将其评分为相关。
        给出一个二进制分数 'yes' 或 'no'，以指示文档是否与问题相关。""",
        input_variables=["context", "question"],
    )


    # 链
    chain = prompt | llm_with_tool


    messages = state["messages"]
    last_message = messages[-1]


    question = messages[0].content
    docs = last_message.content


    scored_result = chain.invoke({"question": question, "context": docs})


    score = scored_result.binary_score


    if score == "yes":
        return "generate"
    else:
        return "rewrite"

def rewrite(state):
    messages = state["messages"]
    question = messages[0].content


    msg = [
        HumanMessage(
            content=f""" 
        查看输入并尝试推理出底层的语义意图/含义。
        这是初始问题：


        -------
        {question} 
        -------
        制定一个改进的问题:""",
        )
    ]


    # LLM
    model = ChatGoogleGenerativeAI(
        model="gemini-1.5-pro",
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )
    response = model.invoke(msg)
    return {"messages": [response]}
#生成最终答案
一旦确认检索到的文档与用户的查询相关，智能体将检索到的文档发送给最终答案生成。


def generate(state):
    messages = state["messages"]
    question = messages[0].content
    last_message = messages[-1]


    docs = last_message.content


    # 提示
    prompt = hub.pull("rlm/rag-prompt")


    # LLM
    llm = ChatGoogleGenerativeAI(
        model="gemini-1.5-pro",
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )


    # 链
    rag_chain = prompt | llm | StrOutputParser()


    # 运行
    response = rag_chain.invoke({"context": docs, "question": question})
    return {"messages": [response]}