@h~1SrSSKJrJr SSKJr SSKJr SSKJ r SSK J r SSK J r SSKJr SS KJr SS KJrJr SS KJr \"S S SS9"SS\ 55rg)zMChain for applying constitutional principles to the outputs of another chain.)AnyOptional) deprecated)CallbackManagerForChainRun)BaseLanguageModel)BasePromptTemplate)Chain)ConstitutionalPrinciple) PRINCIPLES)CRITIQUE_PROMPTREVISION_PROMPTLLMChainz0.2.13zThis class is deprecated and will be removed in langchain 1.0. See API reference for replacement: https://api.python.langchain.com/en/latest/chains/langchain.chains.constitutional_ai.base.ConstitutionalChain.htmlz1.0)sincemessageremovalcL\rSrSr%Sr\\S'\\\S'\\S'\\S'Sr \ \S'\ SS \ \\ S \\4S jj5r\ \\4S \S\S\S\S\S S4 Sjj5r\S \\ 4Sj5r\S \\ 4Sj5rSS\\ \4S\ \S \\ \44Sjjr\S\ S \ 4Sj5rSrg )ConstitutionalChaina7Chain for applying constitutional principles. Note: this class is deprecated. See below for a replacement implementation using LangGraph. The benefits of this implementation are: - Uses LLM tool calling features instead of parsing string responses; - Support for both token-by-token and step-by-step streaming; - Support for checkpointing and memory of chat history; - Easier to modify or extend (e.g., with additional tools, structured responses, etc.) Install LangGraph with: .. code-block:: bash pip install -U langgraph .. code-block:: python from typing import List, Optional, Tuple from langchain.chains.constitutional_ai.prompts import ( CRITIQUE_PROMPT, REVISION_PROMPT, ) from langchain.chains.constitutional_ai.models import ConstitutionalPrinciple from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langgraph.graph import END, START, StateGraph from typing_extensions import Annotated, TypedDict llm = ChatOpenAI(model="gpt-4o-mini") class Critique(TypedDict): """Generate a critique, if needed.""" critique_needed: Annotated[bool, ..., "Whether or not a critique is needed."] critique: Annotated[str, ..., "If needed, the critique."] critique_prompt = ChatPromptTemplate.from_template( "Critique this response according to the critique request. " "If no critique is needed, specify that.\n\n" "Query: {query}\n\n" "Response: {response}\n\n" "Critique request: {critique_request}" ) revision_prompt = ChatPromptTemplate.from_template( "Revise this response according to the critique and reivsion request.\n\n" "Query: {query}\n\n" "Response: {response}\n\n" "Critique request: {critique_request}\n\n" "Critique: {critique}\n\n" "If the critique does not identify anything worth changing, ignore the " "revision request and return 'No revisions needed'. If the critique " "does identify something worth changing, revise the response based on " "the revision request.\n\n" "Revision Request: {revision_request}" ) chain = llm | StrOutputParser() critique_chain = critique_prompt | llm.with_structured_output(Critique) revision_chain = revision_prompt | llm | StrOutputParser() class State(TypedDict): query: str constitutional_principles: List[ConstitutionalPrinciple] initial_response: str critiques_and_revisions: List[Tuple[str, str]] response: str async def generate_response(state: State): """Generate initial response.""" response = await chain.ainvoke(state["query"]) return {"response": response, "initial_response": response} async def critique_and_revise(state: State): """Critique and revise response according to principles.""" critiques_and_revisions = [] response = state["initial_response"] for principle in state["constitutional_principles"]: critique = await critique_chain.ainvoke( { "query": state["query"], "response": response, "critique_request": principle.critique_request, } ) if critique["critique_needed"]: revision = await revision_chain.ainvoke( { "query": state["query"], "response": response, "critique_request": principle.critique_request, "critique": critique["critique"], "revision_request": principle.revision_request, } ) response = revision critiques_and_revisions.append((critique["critique"], revision)) else: critiques_and_revisions.append((critique["critique"], "")) return { "critiques_and_revisions": critiques_and_revisions, "response": response, } graph = StateGraph(State) graph.add_node("generate_response", generate_response) graph.add_node("critique_and_revise", critique_and_revise) graph.add_edge(START, "generate_response") graph.add_edge("generate_response", "critique_and_revise") graph.add_edge("critique_and_revise", END) app = graph.compile() .. code-block:: python constitutional_principles=[ ConstitutionalPrinciple( critique_request="Tell if this answer is good.", revision_request="Give a better answer.", ) ] query = "What is the meaning of life? Answer in 10 words or fewer." async for step in app.astream( {"query": query, "constitutional_principles": constitutional_principles}, stream_mode="values", ): subset = ["initial_response", "critiques_and_revisions", "response"] print({k: v for k, v in step.items() if k in subset}) Example: .. code-block:: python from langchain_community.llms import OpenAI from langchain.chains import LLMChain, ConstitutionalChain from langchain.chains.constitutional_ai.models import ConstitutionalPrinciple llm = OpenAI() qa_prompt = PromptTemplate( template="Q: {question} A:", input_variables=["question"], ) qa_chain = LLMChain(llm=llm, prompt=qa_prompt) constitutional_chain = ConstitutionalChain.from_llm( llm=llm, chain=qa_chain, constitutional_principles=[ ConstitutionalPrinciple( critique_request="Tell if this answer is good.", revision_request="Give a better answer.", ) ], ) constitutional_chain.run(question="What is the meaning of life?") chainconstitutional_principlescritique_chainrevision_chainFreturn_intermediate_stepsNnamesreturncUc[[R"55$UVs/sH n[UPM sn$s snfN)listr values)clsrnames _/var/www/html/shao/venv/lib/python3.13/site-packages/langchain/chains/constitutional_ai/base.pyget_principles"ConstitutionalChain.get_principless: = ))+, ,-23UT 4 U333s;llmcritique_promptrevision_promptkwargsc >[XS9n[XS9nU"SUUUS.UD6$)zCreate a chain from an LLM.)r&prompt)rrrr)r!r&rr'r(r)rrs r#from_llmConstitutionalChain.from_llms="cB!cB ))    c.URR$)z Input keys.)r input_keysselfs r#r1ConstitutionalChain.input_keysszz$$$r/c2UR(a/SQ$S/$)z Output keys.)outputcritiques_and_revisionsinitial_outputr6)rr2s r# output_keysConstitutionalChain.output_keyss  ) )J Jzr/inputs run_managerc 2U=(d [R"5nURR"S0UDSUR S50D6nUnURR R "S0UD6nURSU-S-URSS9 /nURGHPnURRUUURUR S5S9n URU S 9R5n S U R5;aURU S 45 MUR RUUURU UR"UR S 5S 9R5n U nURX45 URSUR$S3S-URSS9 URSU -S-URSS9 URSU -S-URSS9 GMS SU0n UR&(aX\S'X|S'U $)N callbacksoriginalzInitial response:  yellow)textverbosecolorcritique) input_promptoutput_from_modelcritique_requestr> output_stringzno critique neededrevision)rFrGrHrErevision_requestr>z Applying z...greenz Critique: bluezUpdated response: r6r8r7r,)rget_noop_managerrrun get_childr+formaton_textrCrrrH_parse_critiquestriplowerappendrrMr"r) r3r;r< _run_managerresponseinitial_responserFr7constitutional_principle raw_critiquerErL final_outputs r#_callConstitutionalChain._callsY #S&@&Q&Q&S ::>>  ",,Z8 $zz((//9&9 %069LL  #%(,(F(F $ ..22)"*!9!J!J&00< 3L ++*,eg $x~~'77'.."~>**..)"*!9!J!J!!9!J!J&00< /eg  H # * *H+? @   !9!>!> ?sCfL  !   !H,v5  !   )H4v=  ! [)Gf)1(';  ) )-=) *6M2 3r/rJcrSU;aU$URS5SnSU;aURS5SnU$)NzRevision request:rr@)splitrIs r#rU#ConstitutionalChain._parse_critique:sI m 3 %++,?@C ] ")//7:Mr/r,r)__name__ __module__ __qualname____firstlineno____doc__r__annotations__rr rbool classmethodrstrr$r r rrrr-propertyr1r9dictrr_ staticmethodrU__static_attributes__r,r/r#rrsccJ O#$;<<&+t+&*4S "4 % &44 />.=    ,  ,       $%DI%%T#Y=AJS#XJ89J c3h JXssr/rN)rhtypingrrlangchain_core._apirlangchain_core.callbacksrlangchain_core.language_modelsrlangchain_core.promptsrlangchain.chains.baser )langchain.chains.constitutional_ai.modelsr -langchain.chains.constitutional_ai.principlesr *langchain.chains.constitutional_ai.promptsr r langchain.chains.llmrrr,r/r#r{sWS *?<5'MDW)  } g%ggr/