import gradio as gr import os from dotenv import load_dotenv from pathlib import Path from modules.utils import load_processed_meetings, load_prompt_library from src.modules.fed_tools import search_meetings_by_date, FED_TOOLS from src.modules.llm_completions import stream_fed_agent_response from src.config.theme_config import CUSTOM_CSS, GRADIO_THEME, EXAMPLE_QUESTIONS load_dotenv() _FILE_PATH = Path(__file__).parents[1] FOMC_MEETINGS = load_processed_meetings() PROMPT_LIBRARY = load_prompt_library() def convert_history_to_string(history: list) -> str: previous_messages = "" for msg in history: previous_messages += f"{msg['role'].capitalize()}: {msg['content']}\n\n" return previous_messages def respond_for_chat_interface(message: str, history: list, api_key_input: str = ""): """Enhanced response function for gr.ChatInterface with Fed AI Savant capabilities""" api_key = api_key_input.strip() if api_key_input else os.getenv("FIREWORKS_API_KEY", "") if not api_key: yield "❌ Please enter your Fireworks AI API key in the configuration section above." return message_with_history = convert_history_to_string(history) try: for messages in stream_fed_agent_response( message=message, api_key=api_key, prompt_library=PROMPT_LIBRARY, fed_tools=FED_TOOLS, history=message_with_history ): if isinstance(messages, list) and len(messages) > 0: yield messages else: yield str(messages) except Exception as e: error_msg = f"❌ Error: {str(e)}" yield error_msg # Function to create searchable FOMC meetings accordion def create_fomc_meetings_accordion(): """Create searchable accordion for FOMC meetings""" accordions = [] for meeting in FOMC_MEETINGS: title = f"{meeting['date']} - Rate: {meeting['rate_decision']}" content = f""" **Meeting Title:** {meeting['title']} **Rate Decision:** {meeting['rate_decision']} **Summary:** {meeting['summary']} --- *Click to expand for full meeting details* """ accordions.append((title, content)) return accordions def generate_meetings_html(meetings_list): """Generate HTML for meetings list with Fireworks AI purple palette""" if not meetings_list: return '

No meetings available

' html_content = '
' for meeting in meetings_list: date = meeting.get('date', 'Unknown Date') rate_decision = meeting.get('rate', 'N/A') title = meeting.get('title', 'FOMC Meeting') action = meeting.get('action', 'N/A') magnitude = meeting.get('magnitude', 'N/A') forward_guidance = meeting.get('forward_guidance') economic_outlook = meeting.get('economic_outlook') market_impact = meeting.get('market_impact', 'N/A') url = meeting.get('url', '') factors_html = "" key_factors = meeting.get('key_economic_factors', []) if key_factors and len(key_factors) > 0: factors_html = "

Key Factors:

" html_content += f"""
{date} - {rate_decision}

Action: {action}

Magnitude: {magnitude}

Forward Guidance:

{forward_guidance}

{factors_html}

Economic Outlook:

{economic_outlook}

Market Impact:

{market_impact}

{f'

Source: Fed Minutes PDF

' if url else ''}
""" html_content += '
' return html_content def search_and_format_meetings(query: str): """Search meetings and format them for HTML display""" if not query.strip(): return generate_meetings_html(FOMC_MEETINGS) search_result = search_meetings_by_date(query) if search_result.get("success") and search_result.get("results"): return generate_meetings_html(search_result["results"]) else: return '

No meetings found matching your search.

' with gr.Blocks(css=CUSTOM_CSS, title="Fed AI Savant", theme=GRADIO_THEME) as demo: gr.Markdown("""

Fed AI Savant

Intelligent Analysis of Federal Reserve Policy and FOMC Meetings with gpt-oss 120B

""") with gr.Row(): # Sidebar for FOMC Meetings with gr.Column(scale=1, min_width=300, elem_classes="meetings-container"): gr.Markdown("### FOMC Meetings") # Date search date_search = gr.Textbox( placeholder="Search by date...", label="Search", lines=1, container=False, elem_classes="search-box" ) meetings_accordion = gr.HTML(generate_meetings_html(FOMC_MEETINGS)) # Main content area with gr.Column(scale=2): # Compact header with Fireworks AI logo and API key with gr.Row(elem_classes="compact-header"): with gr.Column(scale=1, min_width=150): gr.Markdown("

Powered by

") gr.Image( value=str(_FILE_PATH / "assets" / "fireworks_logo.png"), height=35, width=140, show_label=False, show_download_button=False, container=False, show_fullscreen_button=False, show_share_button=False, ) with gr.Column(scale=2): val = os.getenv("FIREWORKS_API_KEY", "") api_key_value = gr.Textbox( label="API Key", type="password", placeholder="Enter your Fireworks AI API key", value=val, container=True, elem_classes="compact-input" ) # Example Questions - compact version with gr.Row(elem_classes="compact-examples"): example_1 = gr.Button(EXAMPLE_QUESTIONS[0], size="sm", scale=1) example_2 = gr.Button(EXAMPLE_QUESTIONS[1], size="sm", scale=1) example_3 = gr.Button(EXAMPLE_QUESTIONS[2], size="sm", scale=1) example_4 = gr.Button(EXAMPLE_QUESTIONS[3], size="sm", scale=1) # Chat interface with more space chat_interface = gr.ChatInterface( fn=respond_for_chat_interface, type="messages", chatbot=gr.Chatbot(height=450, show_label=False, type="messages", elem_classes="chat-window"), textbox=gr.Textbox( placeholder="Ask about Fed policy, rate decisions, or FOMC meetings...", scale=10 ), additional_inputs=[api_key_value], cache_examples=False, submit_btn="Send" ) date_search.change( search_and_format_meetings, inputs=date_search, outputs=meetings_accordion ) def set_example_text(text): return text example_1.click( lambda: EXAMPLE_QUESTIONS[0], outputs=chat_interface.textbox ) example_2.click( lambda: EXAMPLE_QUESTIONS[1], outputs=chat_interface.textbox ) example_3.click( lambda: EXAMPLE_QUESTIONS[2], outputs=chat_interface.textbox ) example_4.click( lambda: EXAMPLE_QUESTIONS[3], outputs=chat_interface.textbox ) if __name__ == "__main__": demo.launch()