o JjgR@sdZddlmZddlmZmZmZmZmZm Z ddl m Z ddl m Z mZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlm Z e ddddGddde Z!Gdddeee"e"fZ#dS)z+Base classes for LLM-powered router chains.) annotations)AnyDictListOptionalTypecast) deprecated)AsyncCallbackManagerForChainRunCallbackManagerForChainRun)OutputParserException)BaseLanguageModel)BaseOutputParser)BasePromptTemplate)parse_and_check_json_markdown)model_validator)SelfLLMChain) RouterChainz0.2.12z1.0zUse RunnableLambda to select from multiple prompt templates. See example in API reference: https://api.python.langchain.com/en/latest/chains/langchain.chains.router.llm_router.LLMRouterChain.html)sinceremovalmessagecszeZdZUdZded< eddd#dd Zed$d d Zd%fdd Z d&d'ddZ d&d(ddZ e d)d!d"Z ZS)*LLMRouterChaina A router chain that uses an LLM chain to perform routing. This class is deprecated. See below for a replacement, which offers several benefits, including streaming and batch support. Below is an example implementation: .. code-block:: python from operator import itemgetter from typing import Literal from typing_extensions import TypedDict from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableLambda, RunnablePassthrough from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") prompt_1 = ChatPromptTemplate.from_messages( [ ("system", "You are an expert on animals."), ("human", "{query}"), ] ) prompt_2 = ChatPromptTemplate.from_messages( [ ("system", "You are an expert on vegetables."), ("human", "{query}"), ] ) chain_1 = prompt_1 | llm | StrOutputParser() chain_2 = prompt_2 | llm | StrOutputParser() route_system = "Route the user's query to either the animal or vegetable expert." route_prompt = ChatPromptTemplate.from_messages( [ ("system", route_system), ("human", "{query}"), ] ) class RouteQuery(TypedDict): """Route query to destination.""" destination: Literal["animal", "vegetable"] route_chain = ( route_prompt | llm.with_structured_output(RouteQuery) | itemgetter("destination") ) chain = { "destination": route_chain, # "animal" or "vegetable" "query": lambda x: x["query"], # pass through input query } | RunnableLambda( # if animal, chain_1. otherwise, chain_2. lambda x: chain_1 if x["destination"] == "animal" else chain_2, ) chain.invoke({"query": "what color are carrots"}) r llm_chainafter)modereturnrcCs|jj}|jdur td|S)NzLLMRouterChain requires base llm_chain prompt to have an output parser that converts LLM text output to a dictionary with keys 'destination' and 'next_inputs'. Received a prompt with no output parser.)rprompt output_parser ValueError)selfrr"Z/var/www/html/zoom/venv/lib/python3.10/site-packages/langchain/chains/router/llm_router.pyvalidate_prompths  zLLMRouterChain.validate_prompt List[str]cCs|jjS)zTWill be whatever keys the LLM chain prompt expects. :meta private: )r input_keys)r!r"r"r#r&tszLLMRouterChain.input_keysoutputsDict[str, Any]Nonecs"t|t|dtstdS)N next_inputs)super_validate_outputs isinstancedictr )r!r' __class__r"r#r,|s z LLMRouterChain._validate_outputsNinputs run_manager$Optional[CallbackManagerForChainRun]cCsL|pt}|}|jjdd|i|}ttttf|jj j |}|SN callbacksr") r get_noop_manager get_childrpredictrrstrrrrparse)r!r1r2 _run_managerr5 predictionoutputr"r"r#_calls  zLLMRouterChain._call)Optional[AsyncCallbackManagerForChainRun]csD|pt}|}ttttf|jjdd|i|IdH}|Sr4) r r6r7rrr9rrapredict_and_parse)r!r1r2r;r5r=r"r"r#_acalls  zLLMRouterChain._acallllmr rrkwargsrcKst||d}|dd|i|S)zConvenience constructor.)rBrrNr"r)clsrBrrCrr"r"r#from_llms zLLMRouterChain.from_llm)rr)rr%)r'r(rr))N)r1r(r2r3rr()r1r(r2r?rr()rBr rrrCrrr)__name__ __module__ __qualname____doc____annotations__rr$propertyr&r,r>rA classmethodrE __classcell__r"r"r/r#rs  C     rc@s@eZdZUdZdZded<eZded<dZded<dd d Z dS)RouterOutputParserzs,