>'h4ddlmZddlZddlmZmZ ddlmZddl m Z ddl m Z m Z ddlmZmZdd lmZdd lmZdd lmZdd lmZmZdd lmZdZGddZGddeZGddeZGddeZGdde Zy#e$r ddlmZY|wxYw)) annotationsN)AnyProtocol)Self)meta) BaseModelValidationError) DefaultCache RenderCache)config)env) AsyncError) ChatMessagechat_message_from_text)generate_canary_word0ceZdZdddddd d dZd dZed dZeddZeddZeddZ edd Z dd Z y) BasePromptN)nameversionmetadata canary_word render_cachec|xsi|_|xsttj|_||_|xs t |_tj||_ |xst|_ d|xs ti|_y)aD Prompt constructor. Parameters: text: The template text. name: The name to identify this prompt. version: The version string attached to this prompt. metadata: A key-value set of metadata pairs attached to this prompt. canary_word: The string to use for the `{{canary_word}}` extension. If `None`, a default string will be generated. render_cache: The caching backend to store rendered prompts. If `None`, the default in-memory backend will be used. rN) _metadatastruuiduuid4_name_rawr _render_cacher from_string _templateDEFAULT_VERSION_versionrdefaults)selftextrrrrrs S/home/kushmeetdev/Regenta/Chatbot/venv/lib/python3.12/site-packages/banks/prompt.py__init__zBasePrompt.__init__sm."R.S.  );\^.2? & (M7K7MN c<| |jS||jzSNr')r(datas r* _get_contextzBasePrompt._get_context<s <== dmm##r,c|jSr.)rr(s r*rzBasePrompt.metadataAs ~~r,c|jSr.)r r3s r*rzBasePrompt.nameEs zzr,c|jS)z#Returns the raw text of the prompt.)r!r3s r*rawzBasePrompt.rawIsyyr,c|jSr.)r&r3s r*rzBasePrompt.versionNs }}r,cjtj|j}tj|Sr.)rparser6rfind_undeclared_variables)r(asts r* variableszBasePrompt.variablesRs%ii!--c22r,c$|jd|vS)z^Returns whether the canary word is present in `text`, signalling the prompt might have leaked.rr/)r(r)s r* canary_leakedzBasePrompt.canary_leakedWs}}]+t33r,)r)rr str | Nonerr?rdict[str, Any] | Nonerr?rzRenderCache | NonereturnNone)r0z dict | NonerAdict)rAzdict[str, Any])rAr)rAr?)rAzset[str])r)rrAbool) __name__ __module__ __qualname__r+r1propertyrrr6rr<r>r,r*rrs  "*."&+/OO O  O ( O O)O O@$ 334r,rc$eZdZdZdddZdddZy)Prompta The `Prompt` class is the only thing you need to know about Banks on the Python side. The class can be initialized with a string variable containing the prompt template text, then you can invoke the method `text` on your instance to pass the data needed to render the template and get back the final prompt. A quick example: ```py from banks import Prompt p = Prompt("Write a 500-word blog post on {{ topic }}.") my_topic = "retrogame computing" print(p.text({"topic": my_topic})) ``` Nc|j|}|jj|}|r|S|jj |}|jj |||S) Render the prompt using variables present in `data` Parameters: data: A dictionary containing the context variables. )r1r"getr$rendersetr(r0cachedrendereds r*r)z Prompt.textmsa  &##''- M--d3 tX.r,c|j|}|jj|}|s7|jj |}|jj ||g}|j jdD]'} |jtj|)|s|jtd||S#t$rYUwxYw)rM user)rolecontent) r1r"rNr$rOrPstripsplitappendrmodel_validate_jsonr r)r(r0rSmessageslines r* chat_messageszPrompt.chat_messages}s  &%%))$/~~,,T2H    " "4 2&(NN$**40 D  ? ? EF  OO2Q R#  s $C CCr.r0r@rAr)r0r@rAzlist[ChatMessage])rErFrG__doc__r)r_rIr,r*rKrK\s  r,rKc.eZdZdZdfd ZdddZxZS) AsyncPrompta- Banks provides async support through the machinery [provided by Jinja](https://jinja.palletsprojects.com/en/3.0.x/api/#async-support) Since the Jinja environment is a global state in banks, the library can work either with or without async support, and this must be known before importing anything. If the application using banks runs within an `asyncio` loop, you can do two things to optimize banks' execution: 1. Set the environment variable `BANKS_ASYNC_ENABLED=true`. 2. Use the `AsyncPrompt` class that has an awaitable `run` method. For example, let's render a prompt that contains some calls to the `generate` extension. Those calls will be heavily I/O bound, so other tasks can take advantage and advance while the prompt is being rendered. Example: ```python # Enable async support before importing from banks import os os.environ["BANKS_ASYNC_ENABLED"] = "true" # Show logs to see what's happening at runtime import logging logging.basicConfig(level=logging.INFO) import asyncio from banks import AsyncPrompt prompt_template = """ Generate a tweet about the topic '{{ topic }}' with a positive sentiment. """ async def task(task_id: int, sleep_time: int): logging.info(f"Task {task_id} is running.") await asyncio.sleep(sleep_time) logging.info(f"Task {task_id} done.") async def main(): p = AsyncPrompt(prompt_template) # Schedule the prompt rendering along with two executions of 'task', one sleeping for 10 seconds # and one sleeping for 1 second results = await asyncio.gather(p.text({"topic": "AI frameworks"}), task(1, 10), task(2, 1)) print("All tasks done, rendered prompt:") print(results[0]) asyncio.run(main()) ``` c^t||i|tjs d}t |y)NzaAsync is not enabled. Please set the environment variable 'BANKS_ASYNC_ENABLED=on' and try again.)superr+r ASYNC_ENABLEDr)r(argskwargsmsg __class__s r*r+zAsyncPrompt.__init__s2 $)&)##uCS/ !$r,cK|j|}|jj|}|r|S|jj |d{}|jj |||S7"w)rMN)r1r"rNr$ render_asyncrPrQs r*r)zAsyncPrompt.textsn  &##''- M"nn99$?? tX.@sAA6A4#A6)rgrrhrrArBr.r`)rErFrGrar+r) __classcell__)rjs@r*rcrcs4l"r,rcc,eZdZdZddddZddd dZy) PromptRegistryz:Interface to be implemented by concrete prompt registries.N)rcyr.rI)r(rrs r*rNzPromptRegistry.getr,F) overwritecyr.rI)r(promptrrs r*rPzPromptRegistry.setrqr,)rrrr?rArK)rtrKrrrDrArB)rErFrGrarNrPrIr,r*rorosD6:J7<Jr,rocXeZdZUdZded<dZded<dZded<dZded <ed d Z y) PromptModelz(Serializable representation of a Prompt.rr)Nr?rrr@rcj||j|j|j|jS)N)r)rrr)r6rrr)clsrts r* from_promptzPromptModel.from_prompts& fnnW]WfWfggr,)rxz type[Self]rtrKrAr) rErFrGra__annotations__rrr classmethodryrIr,r*rvrvs=2 ID*GZ&*H#*hhr,rv) __future__rrtypingrrr ImportErrortyping_extensionsjinja2rpydanticrr cacher r r rerrorsrtypesrrutilsrr%rrKrcrorvrIr,r*rs# '/,6'>4>4B;Z;|L*L^KXK h) hM'&'sB BB