""" LangChain + LangGraph 集成示例 数据源:docs/菌群知识库/_chunks.jsonl(22,781 个 chunk) 两种用法: 1. RAG 检索增强(推荐 Qdrant/Pinecone) 2. LangGraph 多步 Agent(带分类路由) """ # ============================================================ # 方式一:RAG 检索增强(最小路径) # ============================================================ from langchain_community.document_loaders import JSONLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings import OpenAIEmbeddings # 或 BGE / text-embedding-3-large from langchain_community.vectorstores import QdrantVectorStore from langchain_community.vectorstores.utils import DistanceStrategy from langchain_core.runnables import RunnableParallel, RunnableLambda # 1. 加载 JSONL loader = JSONLoader( file_path=r"C:\code\cfc\docs\菌群知识库\_chunks.jsonl", jq_schema=".[] | {page_content: .content, metadata: .metadata}", content_key="page_content", metadata_key="metadata", ) docs = loader.load() # 2. Embedding — 中文推荐 bge-m3 或 text-embedding-3-large embeddings = OpenAIEmbeddings(model="text-embedding-3-large") # 3. 向量化存储(Qdrant 本地示例) vector_store = QdrantVectorStore.from_documents( documents=docs, embedding=embeddings, collection_name="guhejk_kb", url="http://localhost:6333", distance=DistanceStrategy.COSINE, ) # 4. RAG 检索器(支持 metadata 过滤) retriever = vector_store.as_retriever( search_type="mmr", search_kwargs={ "k": 5, "filter": { "must": [{"key": "category", "match": {"value": "科普"}}] # 可选:限定分类 } } ) # 5. 简单问答链 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_messages([ ("system", "你是菌群健康领域的专家助手。请根据以下资料回答用户问题,引用具体来源。"), ("human", "问题:{question}\n\n参考资料:{context}"), ]) llm = ChatOpenAI(model="gpt-4o", temperature=0) rag_chain = prompt | llm # 使用 result = rag_chain.invoke({ "question": "肠道菌群和抑郁有什么关系?", "context": retriever.invoke("肠道菌群 抑郁") }) print(result.content) # ============================================================ # 方式二:LangGraph 多步 Agent(带分类路由 + 总结) # ============================================================ from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): question: str category: str # 自动推断的分类 retrieved_chunks: list # 检索到的 chunk summary: str final_answer: str # 节点 1:分类路由 def classify_question(state: AgentState) -> AgentState: """根据用户问题自动推断应检索哪个分类""" q = state["question"] routing = { ("检测", "报告", "测序"): "技术服务", ("临床", "患者", "诊断"): "科普", ("技术", "方法", "测序"): "技术", ("公告", "新闻"): "公告", ("报告解读"): "报告解读", ("视频"): "视频", } for kws, cat in routing.items(): if all(kw in q for kw in kws): return {"category": cat} return {"category": "科普"} # 默认科普 # 节点 2:检索 def retrieve(state: AgentState) -> AgentState: cat = state.get("category", "科普") chunks = retriever.invoke(state["question"]) return {"retrieved_chunks": chunks} # 节点 3:总结 def summarize(state: AgentState) -> AgentState: texts = "\n\n".join([c.page_content for c in state["retrieved_chunks"]]) prompt_summary = ChatPromptTemplate.from_messages([ ("system", "你是一个科学文章总结助手。请总结以下内容,突出关键发现和数据。"), ("human", "{texts}"), ]) summary = (prompt_summary | ChatOpenAI(model="gpt-4o-mini", temperature=0)).invoke({"texts": texts}) return {"summary": summary.content} # 节点 4:生成最终回答 def answer(state: AgentState) -> AgentState: ctx = "\n\n---\n\n".join([c.page_content for c in state["retrieved_chunks"]]) final = rag_chain.invoke({"question": state["question"], "context": ctx}) return {"final_answer": final.content} # 构建图 graph = StateGraph(AgentState) graph.add_node("classify", classify_question) graph.add_node("retrieve", retrieve) graph.add_node("summarize", summarize) graph.add_node("answer", answer) graph.add_edge("classify", "retrieve") graph.add_edge("retrieve", "summarize") graph.add_edge("summarize", "answer") graph.set_entry_point("classify") graph.set_finish_point("answer") memory = MemorySaver() app = graph.compile(checkpointer=memory) # 使用 config = {"configurable": {"thread_id": "user-001"}} result = app.invoke( {"question": "肠道菌群和抑郁有什么关系?宏基因组检测能发现什么?"}, config=config, ) print(result["final_answer"]) # ============================================================ # 方式三:在 cfc-backend (Java) 中调用 LangGraph API # ============================================================ """ // 假设 LangGraph 服务运行在 localhost:8123 POST /invoke { "question": "肠道菌群与糖尿病的关系", "category": "科普" } Response: { "final_answer": "...", "summary": "...", "sources": [ {"postId": "1234", "title": "...", "category": "科普"}, ... ] } """ # ============================================================ # 常用搜索示例 # ============================================================ # 按分类过滤检索 retriever_science = vector_store.as_retriever( search_kwargs={"filter": {"must": [{"key": "category", "match": {"value": "科普"}}]}} ) # 按主题过滤(topics metadata) retriever_depression = vector_store.as_retriever( search_kwargs={"filter": {"must": [{"key": "topics", "match": {"any": ["抑郁", "焦虑"]}}]}} ) # 多路检索合并 retrievers = RunnableParallel({ "science": retriever_science, "clinical": vector_store.as_retriever(search_kwargs={"filter": {"must": [{"key": "topics", "match": {"any": ["临床", "诊断"]}}]}}), }) merged = retrievers | (lambda x: sorted(set([c for chunk_list in x.values() for c in chunk_list]), key=lambda c: c.metadata["pub_date"] or ""))