from typing import List, Dict, Union, Any from uuid import uuid4 from langchain_core.documents import Document from ask_candid.retrieval.sources import ( candid_blog, candid_help, candid_learning, issuelab, youtube ) def html_format_doc(doc: Dict[str, Any], source: str, show_chunks=False) -> str: height_px = 200 html = "" if source == "news": # html = news.article_card_html(doc, height_px, show_chunks) pass elif source == "transactions": pass elif source == "organizations": pass elif source == "issuelab": html = issuelab.issuelab_card_html(doc, height_px, show_chunks) elif source == "youtube": html = youtube.build_card_html(doc, 400, show_chunks) elif source == "candid_blog": html = candid_blog.build_card_html(doc, height_px, show_chunks) elif source == "candid_learning": html = candid_learning.build_card_html(doc, height_px, show_chunks) elif source == "candid_help": html = candid_help.build_card_html(doc, height_px, show_chunks) return html def html_format_docs_chat(docs: List[Document]) -> str: """Formats Candid sources Parameters ---------- docs : List[Document] Retrieved documents for context Returns ------- str Formatted HTML """ html = "" if docs: docs_html = [] for doc in docs: s_name = doc.metadata.get("source", "Source") s_url = doc.metadata.get("url", "URL") s_html = ( "" f"" f"{doc.metadata['title']} | {s_name}" ) docs_html.append(s_html) html = f"

Candid Resources

{'
'.join(docs_html)}
" return html def format_chat_ag_response(chatbot: List[Any]) -> List[Any]: """If we called retriever, we appended sources as as one more message. Here we concatinate HTML of sources with the AI response Returns: _type_: updated chatbot message as HTML """ sources = "" if chatbot: title = (chatbot[-1].get("metadata") or {}).get("title", None) if title == "Sources HTML": sources = chatbot[-1]["content"] chatbot.pop(-1) chatbot[-1]["content"] = chatbot[-1]["content"] + sources return chatbot def valid_inputs(*args) -> bool: return any(a is not None or (isinstance(a, str) and a.strip() != '') for a in args) def get_session_id(thread_id: Union[str, None]) -> str: if not thread_id: thread_id = uuid4().hex return thread_id