从可视化到代码
LangChain 实战进阶

from langchain_ollama import ChatOllama

# 连接本地模型
llm = ChatOllama(model="qwen2.5:7b", base_url="http://localhost:11434")

# 方式1：一次性返回
response = llm.invoke("什么是网站备案？")
print(response.content)

# 方式2：流式输出（逐字返回）
for chunk in llm.stream("网站备案需要什么材料？"):
    print(chunk.content, end="", flush=True)

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# CREATE 框架代码版
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是{role}。
领域：{domain}
职责：{responsibility}
语气：{tone}
最多{max_words}字"""),
    ("human", "{question}"),
])

# LCEL 链式语法
chain = prompt | llm | StrOutputParser()
result = chain.invoke({
    "role": "网站备案专家",
    "domain": "工信部网站备案",
    "question": "备案被退回怎么办？",
    ...
})

step1 = prompt.invoke({...})     # → ChatPromptValue
step2 = llm.invoke(step1)        # → AIMessage
step3 = parser.invoke(step2)     # → 纯文本字符串

# 等价于一步到位：
chain = prompt | llm | parser
result = chain.invoke({...})

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载 PDF
pages = PyPDFLoader("网站备案办事指南.pdf").load()

# 递归切分
splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,      # 每块约200字
    chunk_overlap=30,    # 相邻块重叠30字（防截断）
    separators=["\n\n", "\n", "。", "；", "，", " ", ""],
)
splits = splitter.split_documents(pages)
# 结果：1页PDF → 5个知识块

from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import FAISS

# 嵌入模型
embeddings = OllamaEmbeddings(model="bge-m3")

# 存入 FAISS 向量数据库
vectorstore = FAISS.from_documents(splits, embeddings)

# 相似度检索
results = vectorstore.similarity_search("备案需要什么材料？", k=3)

# 带分数检索（越低越相关）
results = vectorstore.similarity_search_with_score("备案被退回", k=3)
# 结果：score 0.76 → 0.80 → 0.85

# MMR 多样性检索
results = vectorstore.max_marginal_relevance_search("备案流程", k=3)

# RAG 提示词（防幻觉核心）
rag_prompt = ChatPromptTemplate.from_template("""你是网站备案专家。根据以下资料回答问题。

【资料】{context}
【规范】只根据资料回答，不编造。无相关信息时说明"根据现有资料无法回答"。
【问题】{question}""")

# 组装 RAG 链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt | llm | StrOutputParser()
)

from langchain_core.tools import tool

# 定义工具
@tool
def search_filing_info(question: str) -> str:
    """搜索网站备案相关信息"""
    return retriever.invoke(question)

# 绑定工具到 LLM
llm_with_tools = llm.bind_tools([search_filing_info, calculate_days])

# Agent 执行循环
def run_agent(question):
    messages = [HumanMessage(content=question)]
    while True:
        response = llm_with_tools.invoke(messages)
        if response.tool_calls:
            # LLM 要调工具 → 执行工具 → 结果加入对话
            ...
        else:
            return response.content  # 不需要工具 → 直接回答

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="网站备案 RAG API", version="1.0.0")

class AskRequest(BaseModel):
    question: str

@app.post("/ask")
async def ask(request: AskRequest):
    answer = rag_chain.invoke(request.question)
    return {"answer": answer, "question": request.question}

@app.post("/ask/stream")
async def ask_stream(request: AskRequest):
    def generate():
        for chunk in rag_chain.stream(request.question):
            yield f"data: {chunk}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

import streamlit as st

# 侧边栏：上传文件
uploaded_files = st.sidebar.file_uploader("上传PDF/TXT", type=["pdf", "txt"])

# 聊天界面
if prompt := st.chat_input("输入你的问题..."):
    with st.chat_message("user"):
        st.write(prompt)
    with st.chat_message("assistant"):
        # 流式输出
        for chunk in rag_chain.stream(prompt):
            st.write(chunk, end="")

from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    question: str
    documents: list
    answer: str

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("grade", grade_documents_node)
workflow.add_node("generate", generate_node)
workflow.add_node("search", search_tool_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", route_after_grading, {
    "generate": "generate",  # 有文档 → 生成
    "search": "search",      # 无文档 → 搜索
})
workflow.add_edge("generate", END)
workflow.add_edge("search", END)

app_graph = workflow.compile()

# LlamaIndex：1步搞定索引（自动切分+向量化）
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=3)
response = query_engine.query("网站备案需要什么材料？")

# 对比 LangChain：需要3步
# splits = splitter.split_documents(pages)
# vectorstore = FAISS.from_documents(splits, embeddings)
# retriever = vectorstore.as_retriever()

┌──────────────────────────┐
│  [Title] 网站备案办事指南  │ ← 识别为标题
├──────────────────────────┤
│  [NarrativeText]          │ ← 识别为正文
│  第一步：备案单位...       │
├──────────────────────────┤
│  [Table]                  │ ← 识别为表格（不切断）
│  材料名称 | 材料形式      │
├──────────────────────────┤
│  [ListItem]               │ ← 识别为列表项
│  1. 登录系统              │
└──────────────────────────┘