🧠 Agentic RAG

# CRAG 的 5 个节点
workflow.add_node("retrieve", retrieve)              # 检索文档
workflow.add_node("grade_documents", grade_documents) # ★ 评分文档相关性
workflow.add_node("generate", generate)              # 生成答案
workflow.add_node("transform_query", transform_query) # 改写 query
workflow.add_node("web_search_node", web_search)     # web 搜索回退

# 关键条件边：评分后决定走哪
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,   # :473 路由函数
    {
        "transform_query": "transform_query",  # 不相关→改写
        "generate": "generate",                # 相关→生成
    },
)
# 改写后重新检索（形成反思循环）
workflow.add_edge("transform_query", "retrieve")

def grade_documents(state):
    """★ 核心反思：用 LLM 逐文档判断是否相关。"""
    filtered_docs = []
    for doc in state["documents"]:
        # 用结构化输出让 LLM 判断 yes/no
        score = retrieval_grader.invoke({"question": q, "document": doc})
        if score.binary_score == "yes":
            filtered_docs.append(doc)   # 保留相关文档
    return {"documents": filtered_docs}

def decide_to_generate(state):
    """★ 路由：相关文档够不够？"""
    if state["documents"]:         # 有相关文档
        return "generate"          # → 直接生成
    return "transform_query"       # 无相关文档 → 改写 query 重试

# pip install langgraph langchain-openai
# export OPENAI_API_KEY="sk-..."
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")

# ============ ① 状态 ============
class State(TypedDict):
    question: str
    documents: Annotated[list, operator.add]   # 检索到的文档（累积）
    generation: str                             # 生成的答案
    web_search_needed: bool                     # 是否需要 web 回退

# ============ ② 检索节点（假装检索，实际接向量库）============
def retrieve(state: State) -> dict:
    """检索文档。这里用假数据模拟。"""
    fake_docs = ["文档A：关于LangChain的内容...", "文档B：无关内容..."]
    return {"documents": [{"content": d} for d in fake_docs]}

# ============ ③ ★ 评分节点（核心反思）============
from pydantic import BaseModel, Field

class Grade(BaseModel):
    """文档相关性评分。"""
    binary_score: str = Field(description="yes 或 no")

grader = model.with_structured_output(Grade)

def grade_documents(state: State) -> dict:
    """★ 反思核心：逐文档评估相关性，过滤掉不相关的。"""
    question = state["question"]
    filtered = []
    for doc in state["documents"]:
        grade = grader.invoke(
            f"问题：{question}\n文档：{doc['content']}\n这文档与问题相关吗？只回 yes/no"
        )
        if grade.binary_score.lower() == "yes":
            filtered.append(doc)
    # 关键：如果过滤后没有相关文档，标记需要 web 回退
    return {
        "documents": [],  # 清空旧文档（避免重复累积，这里简化处理）
        "web_search_needed": len(filtered) == 0,
    }

# ============ ④ 路由：决定生成还是改写回退 ============
def decide_after_grade(state: State) -> str:
    """评分后的路由决策。"""
    if state.get("web_search_needed"):
        return "transform_query"   # 无相关文档 → 改写重试
    return "generate"              # 有相关文档 → 生成

# ============ ⑤ 改写 query（反思循环的"修正"）============
def transform_query(state: State) -> dict:
    """改写 query 以更好检索。"""
    rewriter = (
        ChatPromptTemplate.from_messages([
            ("system", "改写下面的问题使其更适合检索。只输出改写后的问题。"),
            ("human", "{question}"),
        ])
        | model | StrOutputParser()
    )
    better = rewriter.invoke({"question": state["question"]})
    return {"question": better}   # 用新 query 重试检索

# ============ ⑥ 生成节点 ============
def generate(state: State) -> dict:
    """基于文档生成答案。"""
    context = "\n".join(d["content"] for d in state["documents"]) or "（无相关文档）"
    chain = (
        ChatPromptTemplate.from_messages([
            ("system", "根据上下文回答。{context}"),
            ("human", "{question}"),
        ])
        | model | StrOutputParser()
    )
    answer = chain.invoke({"context": context, "question": state["question"]})
    return {"generation": answer}

# ============ ⑦ 构建 CRAG 图 ============
graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("grade_documents", grade_documents)
graph.add_node("generate", generate)
graph.add_node("transform_query", transform_query)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "grade_documents")
# ★ 条件边：评分后路由
graph.add_conditional_edges("grade_documents", decide_after_grade,
    {"generate": "generate", "transform_query": "transform_query"})
# 改写后回到检索（反思循环！）
graph.add_edge("transform_query", "retrieve")
graph.add_edge("generate", END)

app = graph.compile()

# ============ ⑧ 运行 ============
result = app.invoke({
    "question": "LangChain 的 LCEL 是什么？",
    "documents": [], "generation": "", "web_search_needed": False,
})
print("问题:", result["question"])
print("答案:", result["generation"])
# 流程：retrieve → grade → (不相关?) → transform_query → retrieve → grade → (相关) → generate

# Self-RAG 在 generate 之后加两个验证节点：

# 验证①：幻觉检测——生成的内容是否忠实于文档？
def grade_generation_vs_documents(state):
    """生成 vs 文档：有没有编造（幻觉）？"""
    grade = hallucination_grader.invoke({
        "documents": state["documents"],
        "generation": state["generation"],
    })
    if grade.binary_score == "no":   # 检测到幻觉
        return "not supported"        # → 重新生成

    # 验证②：有用性——生成的内容是否回答了问题？
    grade = answer_grader.invoke({
        "question": state["question"],
        "generation": state["generation"],
    })
    if grade.binary_score == "yes":
        return "useful"               # 有用 → 结束
    return "not useful"               # 没回答问题 → 改写重试

# 条件边：三态路由
graph.add_conditional_edges("generate", grade_generation_vs_documents,
    {"not supported": "generate",      # 幻觉→重生
     "useful": END,                    # 有用→结束
     "not useful": "transform_query"}) # 没用→改写

def route_question(state):
    """问题路由：根据问题类型决定检索方式。"""
    source = question_router.invoke({"question": state["question"]})
    if source == "web_search":
        return "web_search"      # 时事/外部知识 → web
    elif source == "vectorstore":
        return "retrieve"        # 内部文档 → 向量检索
# 然后接 CRAG/Self-RAG 的反思流程

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def retrieve(query: str) -> str:
    """检索知识库。当你需要查找信息时使用。"""
    docs = vectorstore.invoke(query)
    return "\n".join(d.page_content for d in docs)

@tool
def web_search(query: str) -> str:
    """搜索网络。当知识库没有时使用。"""
    ...

# 把检索变成工具，交给 ReAct Agent 自主决策
agent = create_react_agent(model, tools=[retrieve, web_search])
# Agent 自己决定：调几次 retrieve？要不要 web_search？什么时候够了？
# 这就是 OpenCode 的范式（用 grep/glob/read 检索代码）