o |!g@sddlmZddlZddlmZmZzddlmZWney)ddlmZYnwddl m Z ddl m Z m Z ddlmZmZdd lmZdd lmZdd lmZdd lmZmZdd lmZdZGdddZGdddeZGdddeZGdddeZGddde ZdS)) annotationsN)AnyProtocol)Self)meta) BaseModelValidationError) DefaultCache RenderCache)config)env) AsyncError) ChatMessagechat_message_from_text)generate_canary_word0c@s~eZdZddddddd%ddZd&ddZed'ddZed(ddZed(ddZed)ddZ ed*d d!Z d+d#d$Z dS), BasePromptN)nameversionmetadata canary_word render_cachetextstrr str | Nonerrdict[str, Any] | NonerrRenderCache | NonereturnNonecCsX|pi|_|p tt|_||_|pt|_t ||_ |p t |_ d|p't i|_dS)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) _metadataruuiduuid4_name_rawr _render_cacher from_string _templateDEFAULT_VERSION_versionrdefaults)selfrrrrrrr,N/mnt/skqttb/ctump_chatbot/chatbot/lib/python3.10/site-packages/banks/prompt.py__init__s    zBasePrompt.__init__data dict | NonedictcCs|dur|jS||jBSNr*)r+r/r,r,r- _get_context<s zBasePrompt._get_contextdict[str, Any]cC|jSr2)r r+r,r,r-rAzBasePrompt.metadatacCr6r2)r#r7r,r,r-rEr8zBasePrompt.namecCr6)z#Returns the raw text of the prompt.)r$r7r,r,r-rawIszBasePrompt.rawcCr6r2)r)r7r,r,r-rNr8zBasePrompt.versionset[str]cCst|j}t|Sr2)r parser9rfind_undeclared_variables)r+astr,r,r- variablesRs  zBasePrompt.variablesboolcCs|jd|vS)z^Returns whether the canary word is present in `text`, signalling the prompt might have leaked.rr3)r+rr,r,r- canary_leakedWszBasePrompt.canary_leaked)rrrrrrrrrrrrrr)r/r0rr1)rr5)rr)rr)rr:)rrrr?) __name__ __module__ __qualname__r.r4propertyrrr9rr>r@r,r,r,r-rs&      rc@s(eZdZdZd d ddZd dd d ZdS)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})) ``` Nr/rrrcCs<||}|j|}|r|S|j|}|j|||S) Render the prompt using variables present in `data` Parameters: data: A dictionary containing the context variables. )r4r%getr'rendersetr+r/cachedrenderedr,r,r-rms   z Prompt.textlist[ChatMessage]c Cs||}|j|}|s|j|}|j||g}|dD]}z |t |Wq#t y8Yq#w|sD|t d|d|S)rF user)rolecontent) r4r%rGr'rHrIstripsplitappendrmodel_validate_jsonrr)r+r/rLmessagesliner,r,r- chat_messages}s    zPrompt.chat_messagesr2r/rrr)r/rrrM)rArBrC__doc__rrXr,r,r,r-rE\s rEcs.eZdZdZdfdd Zddd dZZS) 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()) ``` argsrkwargsrrcs(tj|i|tjsd}t|dS)NzaAsync is not enabled. Please set the environment variable 'BANKS_ASYNC_ENABLED=on' and try again.)superr.r ASYNC_ENABLEDr)r+r\r]msg __class__r,r-r.s zAsyncPrompt.__init__Nr/rrcsD||}|j|}|r|S|j|IdH}|j|||S)rFN)r4r%rGr' render_asyncrIrJr,r,r-rs  zAsyncPrompt.text)r\rr]rrrr2rY)rArBrCrZr.r __classcell__r,r,rar-r[s6r[c@s0eZdZdZdddd d Zd d dddZdS)PromptRegistryz:Interface to be implemented by concrete prompt registries.N)rrrrrrrEcCdSr2r,)r+rrr,r,r-rGzPromptRegistry.getF) overwritepromptrhr?rcCrfr2r,)r+rirhr,r,r-rIrgzPromptRegistry.set)rrrrrrE)rirErhr?rr)rArBrCrZrGrIr,r,r,r-resrec@sLeZdZUdZded<dZded<dZded<dZded <edddZ dS) PromptModelz(Serializable representation of a Prompt.rrNrrrrrcls type[Self]rirErrcCs||j|j|j|jdS)N)rrrr)r9rrr)rkrir,r,r- from_promptszPromptModel.from_prompt)rkrlrirErr) rArBrCrZ__annotations__rrr classmethodrmr,r,r,r-rjs    rj) __future__rr!typingrrr ImportErrortyping_extensionsjinja2rpydanticrrcacher r r r errorsrtypesrrutilsrr(rrEr[rerjr,r,r,r-s,       A>O