<div id="readability-page-1" class="page">이 글은&nbsp;Langgraph를 이용해, 간단히 다중 LLM 에이전트를 개발하는 방법을 정리한 것이다.<div>from dotenv import load_dotenvimport osload_dotenv()from langchain.globals import set_verbose, set_debugset_debug(True)set_verbose(True)local_llm = 'llama3'from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import WebBaseLoaderfrom langchain_milvus import Milvusfrom langchain_community.embeddings import HuggingFaceEmbeddingsurls = [&nbsp; &nbsp; "https://lilianweng.github.io/posts/2023-06-23-agent/",&nbsp; &nbsp; "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",&nbsp; &nbsp; "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",]docs = [WebBaseLoader(url).load() for url in urls]docs_list = [item for sublist in docs for item in sublist]text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(&nbsp; &nbsp; chunk_size=250, chunk_overlap=0)doc_splits = text_splitter.split_documents(docs_list)# Add to Milvusvectorstore = Milvus.from_documents(&nbsp; &nbsp; documents=doc_splits,&nbsp; &nbsp; collection_name="rag_milvus",&nbsp; &nbsp; embedding=HuggingFaceEmbeddings(),&nbsp; &nbsp; connection_args={"uri": "./milvus_rag.db"},)retriever = vectorstore.as_retriever()from langchain.prompts import PromptTemplatefrom langchain_community.chat_models import ChatOllamafrom langchain_core.output_parsers import JsonOutputParser# LLMllm = ChatOllama(model=local_llm, format="json", temperature=0)prompt = PromptTemplate(&nbsp; &nbsp; template="""You are a grader assessing relevance&nbsp;&nbsp; &nbsp; of a retrieved document to a user question. If the document contains keywords related to the user question,&nbsp;&nbsp; &nbsp; grade it as relevant. It does not need to be a stringent test. The goal is to filter out erroneous retrievals.&nbsp;&nbsp; &nbsp; Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question.&nbsp; &nbsp; Provide the binary score as a JSON with a single key 'score' and no premable or explaination.&nbsp; &nbsp; Here is the retrieved document:&nbsp;&nbsp; &nbsp; {document}&nbsp; &nbsp; Here is the user question:&nbsp;&nbsp; &nbsp; {question}&nbsp; &nbsp; """,&nbsp; &nbsp; input_variables=["question", "document"],)retrieval_grader = prompt | llm | JsonOutputParser()question = "agent memory"docs = retriever.invoke(question)doc_txt = docs[1].page_contentprint(retrieval_grader.invoke({"question": question, "document": doc_txt}))from langchain.prompts import PromptTemplatefrom langchain import hubfrom langchain_core.output_parsers import StrOutputParser# Promptprompt = PromptTemplate(&nbsp; &nbsp; template="""You are an assistant for question-answering tasks.&nbsp;&nbsp; &nbsp; Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know.&nbsp;&nbsp; &nbsp; Use three sentences maximum and keep the answer concise:&nbsp; &nbsp; Question: {question}&nbsp;&nbsp; &nbsp; Context: {context}&nbsp;&nbsp; &nbsp; Answer:&nbsp;&nbsp; &nbsp; """,&nbsp; &nbsp; input_variables=["question", "document"],)llm = ChatOllama(model=local_llm, temperature=0)# Post-processingdef format_docs(docs):&nbsp; &nbsp; return "\n\n".join(doc.page_content for doc in docs)# Chainrag_chain = prompt | llm | StrOutputParser()# Runquestion = "agent memory"docs = retriever.invoke(question)generation = rag_chain.invoke({"context": docs, "question": question})print(generation)# LLMllm = ChatOllama(model=local_llm, format="json", temperature=0)# Promptprompt = PromptTemplate(&nbsp; &nbsp; template="""You are a grader assessing whether&nbsp;&nbsp; &nbsp; an answer is grounded in / supported by a set of facts. Give a binary score 'yes' or 'no' score to indicate&nbsp;&nbsp; &nbsp; whether the answer is grounded in / supported by a set of facts. Provide the binary score as a JSON with a&nbsp;&nbsp; &nbsp; single key 'score' and no preamble or explanation.&nbsp; &nbsp; Here are the facts:&nbsp; &nbsp; {documents}&nbsp;&nbsp; &nbsp; Here is the answer:&nbsp;&nbsp; &nbsp; {generation}&nbsp; &nbsp; """,&nbsp; &nbsp; input_variables=["generation", "documents"],)hallucination_grader = prompt | llm | JsonOutputParser()hallucination_grader.invoke({"documents": docs, "generation": generation})### Answer Grader&nbsp;# LLMllm = ChatOllama(model=local_llm, format="json", temperature=0)# Promptprompt = PromptTemplate(&nbsp; &nbsp; template="""You are a grader assessing whether an&nbsp;&nbsp; &nbsp; answer is useful to resolve a question. Give a binary score 'yes' or 'no' to indicate whether the answer is&nbsp;&nbsp; &nbsp; useful to resolve a question. Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.&nbsp; &nbsp; Here is the answer:&nbsp; &nbsp; {generation}&nbsp;&nbsp; &nbsp; Here is the question: {question}&nbsp; &nbsp; """,&nbsp; &nbsp; input_variables=["generation", "question"],)answer_grader = prompt | llm | JsonOutputParser()answer_grader.invoke({"question": question,"generation": generation})### Routerfrom langchain.prompts import PromptTemplatefrom langchain_community.chat_models import ChatOllamafrom langchain_core.output_parsers import JsonOutputParser# LLMllm = ChatOllama(model=local_llm, format="json", temperature=0)prompt = PromptTemplate(&nbsp; &nbsp; template="""You are an expert at routing a&nbsp;&nbsp; &nbsp; user question to a vectorstore or web search. Use the vectorstore for questions on LLM&nbsp; agents,&nbsp;&nbsp; &nbsp; prompt engineering, and adversarial attacks. You do not need to be stringent with the keywords&nbsp;&nbsp; &nbsp; in the question related to these topics. Otherwise, use web-search. Give a binary choice 'web_search'&nbsp;&nbsp; &nbsp; or 'vectorstore' based on the question. Return the a JSON with a single key 'datasource' and&nbsp;&nbsp; &nbsp; no premable or explaination.&nbsp;&nbsp; &nbsp; Question to route:&nbsp;&nbsp; &nbsp; {question}""",&nbsp; &nbsp; input_variables=["question"],)question_router = prompt | llm | JsonOutputParser()question = "llm agent memory"docs = retriever.get_relevant_documents(question)doc_txt = docs[1].page_contentprint(question_router.invoke({"question": question}))### Searchfrom langchain_community.tools.tavily_search import TavilySearchResultsweb_search_tool = TavilySearchResults(k=3)from typing_extensions import TypedDictfrom typing import List### Stateclass GraphState(TypedDict):&nbsp; &nbsp; """&nbsp; &nbsp; Represents the state of our graph.&nbsp; &nbsp; Attributes:&nbsp; &nbsp; &nbsp; &nbsp; question: question&nbsp; &nbsp; &nbsp; &nbsp; generation: LLM generation&nbsp; &nbsp; &nbsp; &nbsp; web_search: whether to add search&nbsp; &nbsp; &nbsp; &nbsp; documents: list of documents&nbsp;&nbsp; &nbsp; """&nbsp; &nbsp; question : str&nbsp; &nbsp; generation : str&nbsp; &nbsp; web_search : str&nbsp; &nbsp; documents : List[str]from langchain.schema import Document### Nodesdef retrieve(state):&nbsp; &nbsp; """&nbsp; &nbsp; Retrieve documents from vectorstore&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): New key added to state, documents, that contains retrieved documents&nbsp; &nbsp; """&nbsp; &nbsp; print("---RETRIEVE---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; # Retrieval&nbsp; &nbsp; documents = retriever.invoke(question)&nbsp; &nbsp; return {"documents": documents, "question": question}def generate(state):&nbsp; &nbsp; """&nbsp; &nbsp; Generate answer using RAG on retrieved documents&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): New key added to state, generation, that contains LLM generation&nbsp; &nbsp; """&nbsp; &nbsp; print("---GENERATE---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; documents = state["documents"]&nbsp; &nbsp; # RAG generation&nbsp; &nbsp; generation = rag_chain.invoke({"context": documents, "question": question})&nbsp; &nbsp; return {"documents": documents, "question": question, "generation": generation}def grade_documents(state):&nbsp; &nbsp; """&nbsp; &nbsp; Determines whether the retrieved documents are relevant to the question&nbsp; &nbsp; If any document is not relevant, we will set a flag to run web search&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): Filtered out irrelevant documents and updated web_search state&nbsp; &nbsp; """&nbsp; &nbsp; print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; documents = state["documents"]&nbsp; &nbsp; # Score each doc&nbsp; &nbsp; filtered_docs = []&nbsp; &nbsp; web_search = "No"&nbsp; &nbsp; for d in documents:&nbsp; &nbsp; &nbsp; &nbsp; score = retrieval_grader.invoke({"question": question, "document": d.page_content})&nbsp; &nbsp; &nbsp; &nbsp; grade = score['score']&nbsp; &nbsp; &nbsp; &nbsp; # Document relevant&nbsp; &nbsp; &nbsp; &nbsp; if grade.lower() == "yes":&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print("---GRADE: DOCUMENT RELEVANT---")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; filtered_docs.append(d)&nbsp; &nbsp; &nbsp; &nbsp; # Document not relevant&nbsp; &nbsp; &nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print("---GRADE: DOCUMENT NOT RELEVANT---")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # We do not include the document in filtered_docs&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # We set a flag to indicate that we want to run web search&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; web_search = "Yes"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; continue&nbsp; &nbsp; return {"documents": filtered_docs, "question": question, "web_search": web_search}def web_search(state):&nbsp; &nbsp; """&nbsp; &nbsp; Web search based based on the question&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): Appended web results to documents&nbsp; &nbsp; """&nbsp; &nbsp; print("---WEB SEARCH---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; documents = state["documents"]&nbsp; &nbsp; # Web search&nbsp; &nbsp; docs = web_search_tool.invoke({"query": question})&nbsp; &nbsp; web_results = "\n".join([d["content"] for d in docs])&nbsp; &nbsp; web_results = Document(page_content=web_results)&nbsp; &nbsp; if documents is not None:&nbsp; &nbsp; &nbsp; &nbsp; documents.append(web_results)&nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; documents = [web_results]&nbsp; &nbsp; return {"documents": documents, "question": question}### Conditional edgedef route_question(state):&nbsp; &nbsp; """&nbsp; &nbsp; Route question to web search or RAG.&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; str: Next node to call&nbsp; &nbsp; """&nbsp; &nbsp; print("---ROUTE QUESTION---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; print(question)&nbsp; &nbsp; source = question_router.invoke({"question": question})&nbsp;&nbsp;&nbsp; &nbsp; print(source)&nbsp; &nbsp; print(source['datasource'])&nbsp; &nbsp; if source['datasource'] == 'web_search':&nbsp; &nbsp; &nbsp; &nbsp; print("---ROUTE QUESTION TO WEB SEARCH---")&nbsp; &nbsp; &nbsp; &nbsp; return "websearch"&nbsp; &nbsp; elif source['datasource'] == 'vectorstore':&nbsp; &nbsp; &nbsp; &nbsp; print("---ROUTE QUESTION TO RAG---")&nbsp; &nbsp; &nbsp; &nbsp; return "vectorstore"def decide_to_generate(state):&nbsp; &nbsp; """&nbsp; &nbsp; Determines whether to generate an answer, or add web search&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; str: Binary decision for next node to call&nbsp; &nbsp; """&nbsp; &nbsp; print("---ASSESS GRADED DOCUMENTS---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; web_search = state["web_search"]&nbsp; &nbsp; filtered_documents = state["documents"]&nbsp; &nbsp; if web_search == "Yes":&nbsp; &nbsp; &nbsp; &nbsp; # All documents have been filtered check_relevance&nbsp; &nbsp; &nbsp; &nbsp; # We will re-generate a new query&nbsp; &nbsp; &nbsp; &nbsp; print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, INCLUDE WEB SEARCH---")&nbsp; &nbsp; &nbsp; &nbsp; return "websearch"&nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; # We have relevant documents, so generate answer&nbsp; &nbsp; &nbsp; &nbsp; print("---DECISION: GENERATE---")&nbsp; &nbsp; &nbsp; &nbsp; return "generate"### Conditional edgedef grade_generation_v_documents_and_question(state):&nbsp; &nbsp; """&nbsp; &nbsp; Determines whether the generation is grounded in the document and answers question.&nbsp; &nbsp; Args:&nbsp; &nbsp; &nbsp; &nbsp; state (dict): The current graph state&nbsp; &nbsp; Returns:&nbsp; &nbsp; &nbsp; &nbsp; str: Decision for next node to call&nbsp; &nbsp; """&nbsp; &nbsp; print("---CHECK HALLUCINATIONS---")&nbsp; &nbsp; question = state["question"]&nbsp; &nbsp; documents = state["documents"]&nbsp; &nbsp; generation = state["generation"]&nbsp; &nbsp; score = hallucination_grader.invoke({"documents": documents, "generation": generation})&nbsp; &nbsp; grade = score['score']&nbsp; &nbsp; # Check hallucination&nbsp; &nbsp; if grade == "yes":&nbsp; &nbsp; &nbsp; &nbsp; print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")&nbsp; &nbsp; &nbsp; &nbsp; # Check question-answering&nbsp; &nbsp; &nbsp; &nbsp; print("---GRADE GENERATION vs QUESTION---")&nbsp; &nbsp; &nbsp; &nbsp; score = answer_grader.invoke({"question": question,"generation": generation})&nbsp; &nbsp; &nbsp; &nbsp; grade = score['score']&nbsp; &nbsp; &nbsp; &nbsp; if grade == "yes":&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print("---DECISION: GENERATION ADDRESSES QUESTION---")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return "useful"&nbsp; &nbsp; &nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return "not useful"&nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; pprint("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")&nbsp; &nbsp; &nbsp; &nbsp; return "not supported"from langgraph.graph import END, StateGraphworkflow = StateGraph(GraphState)# Define the nodesworkflow.add_node("websearch", web_search) # web searchworkflow.add_node("retrieve", retrieve) # retrieveworkflow.add_node("grade_documents", grade_documents) # grade documentsworkflow.add_node("generate", generate) # generatae# Build graphworkflow.set_conditional_entry_point(&nbsp; &nbsp; route_question,&nbsp; &nbsp; {&nbsp; &nbsp; &nbsp; &nbsp; "websearch": "websearch",&nbsp; &nbsp; &nbsp; &nbsp; "vectorstore": "retrieve",&nbsp; &nbsp; },)workflow.add_edge("retrieve", "grade_documents")workflow.add_conditional_edges(&nbsp; &nbsp; "grade_documents",&nbsp; &nbsp; decide_to_generate,&nbsp; &nbsp; {&nbsp; &nbsp; &nbsp; &nbsp; "websearch": "websearch",&nbsp; &nbsp; &nbsp; &nbsp; "generate": "generate",&nbsp; &nbsp; },)workflow.add_edge("websearch", "generate")workflow.add_conditional_edges(&nbsp; &nbsp; "generate",&nbsp; &nbsp; grade_generation_v_documents_and_question,&nbsp; &nbsp; {&nbsp; &nbsp; &nbsp; &nbsp; "not supported": "generate",&nbsp; &nbsp; &nbsp; &nbsp; "useful": END,&nbsp; &nbsp; &nbsp; &nbsp; "not useful": "websearch",&nbsp; &nbsp; },)# Compileapp = workflow.compile()# Testfrom pprint import pprintinputs = {"question": "What are the types of agent memory?"}for output in app.stream(inputs):&nbsp; &nbsp; for key, value in output.items():&nbsp; &nbsp; &nbsp; &nbsp; pprint(f"Finished running: {key}:")pprint(value["generation"])</div></div>

Daddy Makers

이 글은 Langgraph를 이용해, 간단히 다중 LLM 에이전트를 개발하는 방법을 정리한 것이다.
Local Agentic RAG with LangGraph and Llama
from dotenv import load_dotenv
import os
load_dotenv()
from langchain.globals import set_verbose, set_debug
set_debug(True)
set_verbose(True)
local_llm = 'llama3'
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_milvus import Milvus
from langchain_community.embeddings import HuggingFaceEmbeddings
urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)
# Add to Milvus
vectorstore = Milvus.from_documents(
    documents=doc_splits,
    collection_name="rag_milvus",
    embedding=HuggingFaceEmbeddings(),
    connection_args={"uri": "./milvus_rag.db"},
)
retriever = vectorstore.as_retriever()
from langchain.prompts import PromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import JsonOutputParser
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)
prompt = PromptTemplate(
    template="""You are a grader assessing relevance 
    of a retrieved document to a user question. If the document contains keywords related to the user question, 
    grade it as relevant. It does not need to be a stringent test. The goal is to filter out erroneous retrievals. 
    
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question.
    Provide the binary score as a JSON with a single key 'score' and no premable or explaination.
     
    Here is the retrieved document: 
    {document}
    
    Here is the user question: 
    {question}
    """,
    input_variables=["question", "document"],
)
retrieval_grader = prompt | llm | JsonOutputParser()
question = "agent memory"
docs = retriever.invoke(question)
doc_txt = docs[1].page_content
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))
from langchain.prompts import PromptTemplate
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
# Prompt
prompt = PromptTemplate(
    template="""You are an assistant for question-answering tasks. 
    Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. 
    Use three sentences maximum and keep the answer concise:
    Question: {question} 
    Context: {context} 
    Answer: 
    """,
    input_variables=["question", "document"],
)
llm = ChatOllama(model=local_llm, temperature=0)
# Post-processing
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
# Chain
rag_chain = prompt | llm | StrOutputParser()
# Run
question = "agent memory"
docs = retriever.invoke(question)
generation = rag_chain.invoke({"context": docs, "question": question})
print(generation)
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)
# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether 
    an answer is grounded in / supported by a set of facts. Give a binary score 'yes' or 'no' score to indicate 
    whether the answer is grounded in / supported by a set of facts. Provide the binary score as a JSON with a 
    single key 'score' and no preamble or explanation.
    
    Here are the facts:
    {documents} 
    Here is the answer: 
    {generation}
    """,
    input_variables=["generation", "documents"],
)
hallucination_grader = prompt | llm | JsonOutputParser()
hallucination_grader.invoke({"documents": docs, "generation": generation})
### Answer Grader 
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)
# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether an 
    answer is useful to resolve a question. Give a binary score 'yes' or 'no' to indicate whether the answer is 
    useful to resolve a question. Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.
     
    Here is the answer:
    {generation} 
    Here is the question: {question}
    """,
    input_variables=["generation", "question"],
)
answer_grader = prompt | llm | JsonOutputParser()
answer_grader.invoke({"question": question,"generation": generation})
### Router
from langchain.prompts import PromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import JsonOutputParser
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)
prompt = PromptTemplate(
    template="""You are an expert at routing a 
    user question to a vectorstore or web search. Use the vectorstore for questions on LLM  agents, 
    prompt engineering, and adversarial attacks. You do not need to be stringent with the keywords 
    in the question related to these topics. Otherwise, use web-search. Give a binary choice 'web_search' 
    or 'vectorstore' based on the question. Return the a JSON with a single key 'datasource' and 
    no premable or explaination. 
    
    Question to route: 
    {question}""",
    input_variables=["question"],
)
question_router = prompt | llm | JsonOutputParser()
question = "llm agent memory"
docs = retriever.get_relevant_documents(question)
doc_txt = docs[1].page_content
print(question_router.invoke({"question": question}))
### Search
from langchain_community.tools.tavily_search import TavilySearchResults
web_search_tool = TavilySearchResults(k=3)
from typing_extensions import TypedDict
from typing import List
### State
class GraphState(TypedDict):
    """
    Represents the state of our graph.
    Attributes:
        question: question
        generation: LLM generation
        web_search: whether to add search
        documents: list of documents 
    """
    question : str
    generation : str
    web_search : str
    documents : List[str]
from langchain.schema import Document
### Nodes
def retrieve(state):
    """
    Retrieve documents from vectorstore
    Args:
        state (dict): The current graph state
    Returns:
        state (dict): New key added to state, documents, that contains retrieved documents
    """
    print("---RETRIEVE---")
    question = state["question"]
    # Retrieval
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}
def generate(state):
    """
    Generate answer using RAG on retrieved documents
    Args:
        state (dict): The current graph state
    Returns:
        state (dict): New key added to state, generation, that contains LLM generation
    """
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]
    
    # RAG generation
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}
def grade_documents(state):
    """
    Determines whether the retrieved documents are relevant to the question
    If any document is not relevant, we will set a flag to run web search
    Args:
        state (dict): The current graph state
    Returns:
        state (dict): Filtered out irrelevant documents and updated web_search state
    """
    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]
    
    # Score each doc
    filtered_docs = []
    web_search = "No"
    for d in documents:
        score = retrieval_grader.invoke({"question": question, "document": d.page_content})
        grade = score['score']
        # Document relevant
        if grade.lower() == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        # Document not relevant
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            # We do not include the document in filtered_docs
            # We set a flag to indicate that we want to run web search
            web_search = "Yes"
            continue
    return {"documents": filtered_docs, "question": question, "web_search": web_search}
    
def web_search(state):
    """
    Web search based based on the question
    Args:
        state (dict): The current graph state
    Returns:
        state (dict): Appended web results to documents
    """
    print("---WEB SEARCH---")
    question = state["question"]
    documents = state["documents"]
    # Web search
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    if documents is not None:
        documents.append(web_results)
    else:
        documents = [web_results]
    return {"documents": documents, "question": question}
### Conditional edge
def route_question(state):
    """
    Route question to web search or RAG.
    Args:
        state (dict): The current graph state
    Returns:
        str: Next node to call
    """
    print("---ROUTE QUESTION---")
    question = state["question"]
    print(question)
    source = question_router.invoke({"question": question})  
    print(source)
    print(source['datasource'])
    if source['datasource'] == 'web_search':
        print("---ROUTE QUESTION TO WEB SEARCH---")
        return "websearch"
    elif source['datasource'] == 'vectorstore':
        print("---ROUTE QUESTION TO RAG---")
        return "vectorstore"
def decide_to_generate(state):
    """
    Determines whether to generate an answer, or add web search
    Args:
        state (dict): The current graph state
    Returns:
        str: Binary decision for next node to call
    """
    print("---ASSESS GRADED DOCUMENTS---")
    question = state["question"]
    web_search = state["web_search"]
    filtered_documents = state["documents"]
    if web_search == "Yes":
        # All documents have been filtered check_relevance
        # We will re-generate a new query
        print("---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, INCLUDE WEB SEARCH---")
        return "websearch"
    else:
        # We have relevant documents, so generate answer
        print("---DECISION: GENERATE---")
        return "generate"
### Conditional edge
def grade_generation_v_documents_and_question(state):
    """
    Determines whether the generation is grounded in the document and answers question.
    Args:
        state (dict): The current graph state
    Returns:
        str: Decision for next node to call
    """
    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]
    score = hallucination_grader.invoke({"documents": documents, "generation": generation})
    grade = score['score']
    # Check hallucination
    if grade == "yes":
        print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
        # Check question-answering
        print("---GRADE GENERATION vs QUESTION---")
        score = answer_grader.invoke({"question": question,"generation": generation})
        grade = score['score']
        if grade == "yes":
            print("---DECISION: GENERATION ADDRESSES QUESTION---")
            return "useful"
        else:
            print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
            return "not useful"
    else:
        pprint("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
        return "not supported"
from langgraph.graph import END, StateGraph
workflow = StateGraph(GraphState)
# Define the nodes
workflow.add_node("websearch", web_search) # web search
workflow.add_node("retrieve", retrieve) # retrieve
workflow.add_node("grade_documents", grade_documents) # grade documents
workflow.add_node("generate", generate) # generatae
# Build graph
workflow.set_conditional_entry_point(
    route_question,
    {
        "websearch": "websearch",
        "vectorstore": "retrieve",
    },
)
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "websearch": "websearch",
        "generate": "generate",
    },
)
workflow.add_edge("websearch", "generate")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "websearch",
    },
)
# Compile
app = workflow.compile()
# Test
from pprint import pprint
inputs = {"question": "What are the types of agent memory?"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"Finished running: {key}:")
pprint(value["generation"])
레퍼런스
Integrating LangGraph with Ollama for Advanced LLM Applications | by Aleksandr Lifanov | Medium
Building Local AI Agents: A Guide to LangGraph, AI Agents, and Ollama (paperspace.com)
OllamaFunctions | LangChain
Empowering Human Led Decision Flows with Open Source Model Enabled Function Calls in Langgraph | by Daisha Drayton | Medium
LangGraph + Adaptive Rag + LLama3 Python Project: Easy AI/Chat For Your Docs 2024 (quickaitutorial.com)
LangGraph : Create Your Hyper AI Agent 2024 (quickaitutorial.com)

Langgraph 기반 다중 LLM 에이전트 개발하기

댓글