Untitled
unknown
plain_text
10 months ago
2.4 kB
5
Indexable
import instructor from openai import OpenAI from config import Config from user_prompt_parser import get_user_info from cloudsearch_client import CloudSearchClient, UserPrompt, SearchParameters from customer_search_client import CustomerCloudSearchClient, CustomerSearchParameters from merging_util import merge_child_subsidiary_details import json import concurrent.futures def perform_subsidiary_search(subsidiary_query): cloudsearch_client_subsidiary = CloudSearchClient(Config.CLOUDSEARCH_DOMAIN_ENDPOINT) search_params_subsidiary = SearchParameters(query=subsidiary_query, fields=["subsidiary_full_name"]) return cloudsearch_client_subsidiary.search(params=search_params_subsidiary) def perform_customer_search(vendor_query): cloudsearch_client_vendor = CustomerCloudSearchClient(Config.CUSTOMER_SEARCH_DOMAIN_ENDPOINT) search_params_vendor = CustomerSearchParameters(query=vendor_query) return cloudsearch_client_vendor.search(params=search_params_vendor) def main(): # Initialize OpenAI client openai_client = instructor.from_openai(OpenAI(api_key=Config.OPENAI_API_KEY)) # User message message = "Give me all bills & bill credits for the pre-paid rent account and subsidiary DDE-US for Q2?" # Parse user prompt user_info = get_user_info(openai_client, message) # Prepare search queries subsidiary_query = user_info.subsidiary vendor_query = user_info.vendor # Use ThreadPoolExecutor to handle concurrent searches with concurrent.futures.ThreadPoolExecutor() as executor: future_subsidiary = executor.submit(perform_subsidiary_search, subsidiary_query) future_customer = executor.submit(perform_customer_search, vendor_query) # Retrieve results search_results_subsidiary = future_subsidiary.result() search_results_vendor = future_customer.result() # Merge child subsidiary details merged_details_subsidiary = merge_child_subsidiary_details(search_results_subsidiary) # Print results (or handle them as needed) print("Subsidiary Search Results:") print(search_results_subsidiary.json(indent=2)) print("Merged Subsidiary Details:") print(json.dumps(merged_details_subsidiary, indent=2)) print("Vendor Search Results:") print(search_results_vendor.json(indent=2)) if __name__ == "__main__": main()
Editor is loading...
Leave a Comment