final solution of the 2024 hackathon

app.py

@@ -0,0 +1,110 @@
+import os
+import json
+import gradio as gr
+import json
+from langchain_legal import run_pipeline
+
+global data
+data = {}
+
+
+def load_json(path: str) -> dict:
+    with open(path, 'r') as f:
+        return json.load(f)
+
+
+def get_ai_final_summary(text_input):
+    global data
+    data = run_pipeline(text_input)
+    return data['ai_final_summary']
+
+
+STEP_NAMES = [
+    'Initial assessment',
+    'Applicable EU regs',
+    'Applicable national regs',
+    'Complementary assessment',
+]
+
+
+def get_inital_assesment() -> str:
+    initial = data.get('ai_first_5', None)
+    print(data)
+    if initial is None:
+        return 'No llm answer yet'
+    return initial
+
+
+def get_applicable_eu_regs() -> str:
+    eval = data.get('ai_eur_lex_eval', None)
+    if eval is None:
+        return 'No llm answer yet'
+    sources = data.get('eur_lex_lookup')
+    res = f'EVALUATION:\n{eval}'
+    if sources is not None:
+        res = f'SOURCES:\n{sources}\n\n\n' + res
+    return res
+
+
+def get_applicable_national_regs() -> str:
+    eval = data.get('ai_aus_eval', None)
+    nat = 'aus'
+    if eval is None:
+        eval = data.get('ai_ger_eval', None)
+        nat = 'ger'
+        if eval is None:
+            return 'No llm answer yet'
+    if nat == 'aus':
+        sources = data.get('aus_lookup')
+    else:
+        sources = data.get('ger_lookup')
+    res = f'EVALUATION:\n{eval}'
+    if sources is not None:
+        res = f'SOURCES:\n{sources}\n\n\n' + res
+    return res
+
+
+def get_ai_challange() -> str:
+    challange = data.get('ai_challange', None)
+    print(data)
+    if challange is None:
+        return 'No llm answer yet'
+    return challange
+
+
+CALLBACKS = [
+    get_inital_assesment,
+    get_applicable_eu_regs,
+    get_applicable_national_regs,
+    get_ai_challange,
+]
+
+
+def main():
+    with gr.Blocks() as demo:
+        gr.Markdown("# Lawgarithm's LLM legal research tool")
+
+        with gr.Column():
+            text_input = gr.Textbox(
+                label='Case description',
+                placeholder='Write your legal case here',
+            )
+            submit_btn = gr.Button('Submit')
+
+        text_output = gr.Textbox(label='Final Summary')
+
+        submit_btn.click(
+            get_ai_final_summary, inputs=text_input, outputs=text_output
+        )
+
+        with gr.Tabs():
+            for name, cb in zip(STEP_NAMES, CALLBACKS):
+                with gr.Tab(f'{name}') as t:
+                    tab_text = gr.Textbox(placeholder='No answer yet')
+                    t.select(cb, outputs=tab_text)
+
+    demo.launch()
+
+
+if __name__ == '__main__':
+    main()

langchain_legal.py

@@ -0,0 +1,319 @@
+import os
+import json
+from langchain.chains import ConversationChain
+from langchain.schema import SystemMessage, HumanMessage, AIMessage
+import boto3
+from typing import Dict
+from langchain_aws.retrievers import AmazonKnowledgeBasesRetriever
+from langchain_aws import ChatBedrock
+
+
+def create_bedrock_client(service_type: str = 'bedrock-runtime'):
+    """Create and return a Bedrock client with default credentials"""
+    return boto3.client(
+        service_name=service_type,
+        region_name='eu-west-2',
+    )
+
+
+def initialize_claude_chain(client):
+    """Initialize and return a conversation chain with Claude"""
+    llm = ChatBedrock(
+        model='anthropic.claude-3-sonnet-20240229-v1:0',
+        client=client,
+        model_kwargs={
+            'max_tokens': 1000,
+            'temperature': 0.6,
+            'top_p': 0.9,
+        },
+    )
+    return llm
+
+
+def chat_with_claude(conversation: ConversationChain, user_input: str) -> Dict:
+    """Send a message to Claude and get its response"""
+    try:
+        response = conversation.predict(input=user_input)
+        return {'status': 'success', 'response': response}
+    except Exception as e:
+        return {'status': 'error', 'error': str(e)}
+
+
+def classify_kb_response(llm, response: str) -> list[str]:
+    kb_classifer = [
+        SystemMessage(
+            content="""You are a classifier determining the intent of llm responses.
+
+            The llm can decide to consult 3 databases: ['eur lex', 'austrian law', 'german law']
+            Determine the intent and output the relevant classes.
+            Examples:
+            #1
+            <llm indicates eu law intent>
+            Output:
+            ['eur lex']
+            #2
+            <llm indicates eu law and austrian law intent>
+            Output:
+            ['eur lex', 'austrian law']
+            #3
+            <llm indicates eu law and german law intent>
+            Output:
+            ['eur lex', 'german law']
+            """
+        ),
+        HumanMessage(content=f'Classify the below context:{response}'),
+    ]
+    response = llm.invoke(kb_classifer)
+    res_str = str(response.content)
+    res_parsed = []
+    if 'eur lex' in res_str.lower():
+        res_parsed.append('eur lex')
+    if 'austrian law' in res_str.lower():
+        res_parsed.append('austrian law')
+    if 'german law' in res_str.lower():
+        res_parsed.append('german law')
+    return res_parsed
+
+
+def classify_exception_response(llm, response: str) -> bool:
+    kb_classifer = [
+        SystemMessage(
+            content="""You are a classifier determining the intent of llm responses.
+
+            The llm can decide if the legal case falls upon an exception, and if so
+            can consult a database.
+
+            Determine the intent and output the relevant classes.
+            Examples:
+            #1
+            <llm indicates no exceptions seem to apply confidently>
+            Output:
+            'no exception'
+            #2
+            <llm indicates there might be some exceptions>
+            Output:
+            'database lookup'
+            """
+        ),
+        HumanMessage(content=f'Classify the below context:{response}'),
+    ]
+    response = llm.invoke(kb_classifer)
+    res_str = str(response.content)
+    return 'database lookup' in res_str.lower()
+
+
+def retrieve_knowledge_base_docs(
+    client,
+    knowledge_base_id: str,
+    query: str,
+    num_res: int = 25,
+) -> list:
+    retriever = AmazonKnowledgeBasesRetriever(
+        client=client,
+        knowledge_base_id=knowledge_base_id,
+        retrieval_config={
+            'vectorSearchConfiguration': {'numberOfResults': num_res}
+        },
+        min_score_confidence=0.0,
+    )
+    return retriever.invoke(query)
+
+
+case1 = """A new member is about to be appointed to the management body of a(n ECB) supervised institution in Austria. The
+    prospective member is already a member of the board in three other group companies. The supervisor is concerned the potential board
+    member does not have enough time to commit to this new position. Can the ECB as supervisor oppose this appointment? 
+"""
+
+case2 = """A shareholder in a German bank did not ask approval to the ECB for the
+acquisition of its qualifying holding (10% or more) in the German bank.
+What can be the consequences?
+"""
+
+case3 = """A request for public access to ECB documents based on Decision ECB/2004/3
+(2004/258/EC) aims at getting access to documents containing the proceedings
+and the outcome of the deliberations of a Governing Council meeting.
+Can the documents be disclosed under the applicable legal framework?
+"""
+
+case4 = """In relation to the Corporate sector purchase programme (CSPP):
+Is commercial paper an eligible asset under the CSPP? Under what
+conditions?
+"""
+
+
+SYSTEM = 'You are a lawyer, giving professional and accurate legal advice. Base your answers on relevant, up todate legal frameworks.'
+
+
+def run_pipeline(case_text: str | None = None):
+    client = create_bedrock_client()
+    retrieval_client = create_bedrock_client('bedrock-agent-runtime')
+    chat = initialize_claude_chain(client)
+    export = {}
+    case = ''
+    if case_text is None:
+        case = case1
+    else:
+        case = case_text
+
+    human_input = f"""
+    <case>{case}</case>
+
+    FOLLOW THE STEP BY STEP PLAN:
+    1. Summarize the question
+    2. Identify the main actors and their relationships (if applicable) 
+    3. Identify who is competent on the legal matter and reference the legal basis
+    4. Identify the legal issue based on the inital question. 
+    5. Which knowledge bases are most likely to contain relevant information? (Eur lex, austrian law, german law)
+    """
+    export['human_1'] = human_input
+    main_wf_messages = [
+        SystemMessage(
+            content=SYSTEM,
+        ),
+        HumanMessage(
+            content=human_input,
+        ),
+    ]
+    AUS_KB_ID = 'URGID9GFK8'
+    GER_KB_ID = '19H00QAPZG'
+    EU_KB_ID = 'ORMULTAIWL'
+
+    response = chat.invoke(main_wf_messages)
+    fifth_point = str(response.content).split('5. ')[-1]
+    print(response.content)
+    export['ai_first_5'] = str(response.content)
+    main_wf_messages.append(
+        AIMessage(content=str(response.content)),
+    )
+    print('------')
+    print(fifth_point)
+    print('------')
+    classification = classify_kb_response(chat, fifth_point)
+    export['ai_lookup_decision_1'] = classification
+    print('CLASS', classification)
+    if 'eur lex' in classification:
+        eu_res = retrieve_knowledge_base_docs(
+            retrieval_client, EU_KB_ID, str(response.content)
+        )
+        export['eur_lex_lookup'] = [str(r) for r in eu_res]
+        main_wf_messages.append(
+            HumanMessage(
+                content=f'Determine which regulations apply (from eur lex):\n{eu_res}'
+            )
+        )
+        response = chat.invoke(main_wf_messages)
+        print(response.content)
+        export['ai_eur_lex_eval'] = str(response.content)
+        main_wf_messages.append(
+            AIMessage(content=str(response.content)),
+        )
+    if 'austrian law' in classification:
+        aus_res = retrieve_knowledge_base_docs(
+            retrieval_client,
+            AUS_KB_ID,
+            f'Determine which regulations apply (from austrian law):\n{case}',
+        )
+        export['aus_lookup'] = [str(r) for r in aus_res]
+        main_wf_messages.append(
+            HumanMessage(
+                content=f'Determine which regulations apply (from austrian law):\n{aus_res}'
+            )
+        )
+        response = chat.invoke(main_wf_messages)
+        print(response.content)
+        export['ai_aus_eval'] = str(response.content)
+        main_wf_messages.append(
+            AIMessage(content=str(response.content)),
+        )
+    if 'german law' in classification:
+        ger_res = retrieve_knowledge_base_docs(
+            retrieval_client,
+            GER_KB_ID,
+            f'Determine which regulations apply (from german law):\n{human_input}',
+        )
+        export['ger_lookup'] = [str(r) for r in ger_res]
+        main_wf_messages.append(
+            HumanMessage(
+                content=f'Determine which regulations apply (from german law):\n{ger_res}'
+            )
+        )
+        response = chat.invoke(main_wf_messages)
+        print(response.content)
+        export['ai_ger_eval'] = str(response.content)
+        main_wf_messages.append(
+            AIMessage(content=str(response.content)),
+        )
+    # EXCEPTION_PROMPT = """6. Determine if the case falls upon an exception and
+    #        retrieve the relevant regulation connected to the exception"""
+    # export['human_exception'] = EXCEPTION_PROMPT
+    # main_wf_messages.append(
+    #    HumanMessage(
+    #        content=EXCEPTION_PROMPT,
+    #    )
+    # )
+    # response = chat.invoke(main_wf_messages)
+    # print(response.content)
+    # export['ai_exception_investigation'] = str(response.content)
+    # main_wf_messages.append(
+    #    AIMessage(content=str(response.content)),
+    # )
+    #    is_exception = classify_exception_response(chat, str(response.content))
+    #    print('IS EXCEPTION CLASS:', is_exception)
+    #    if is_exception:
+    #        eu_res = retrieve_knowledge_base_docs(
+    #            retrieval_client,
+    #            EU_KB_ID,
+    #            str(response.content),
+    #            10,
+    #        )
+    #        export['ai_exception_lookup'] = [str(r) for r in eu_res]
+    #        main_wf_messages.append(
+    #            HumanMessage(
+    #                content=f'Determine if the regulations provide a strong legal basis for an exception:\n{eu_res}'
+    #            )
+    #        )
+    #        response = chat.invoke(main_wf_messages)
+    #        print(response.content)
+    #        export['ai_exception_eval'] = str(response.content)
+    #        main_wf_messages.append(
+    #            AIMessage(content=str(response.content)),
+    #        )
+    #
+    FINAL_SUMMARY_PROMPT = """6.Based on the actors, the competencies, the relevant legal frameworks
+        as mentioned above draft an answer to the FULL original question(s). 
+        Mention the legal issue(s), the relevant articles used in both eur lex and national and the
+        applicable regime set up, the conditions if applicable, conclusion."""
+    export['human_summary'] = FINAL_SUMMARY_PROMPT
+    main_wf_messages.append(
+        HumanMessage(
+            content=FINAL_SUMMARY_PROMPT,
+        )
+    )
+    response = chat.invoke(main_wf_messages)
+    print('----------FINAL RESPONSE------------')
+    print(response.content)
+    export['ai_final_summary'] = str(response.content)
+    main_wf_messages.append(
+        AIMessage(content=str(response.content)),
+    )
+
+    main_wf_messages.append(
+        HumanMessage(
+            content="""You are a lawyer and you need explain your answer to the question(s). 
+            Your colleagues are saying you are wrong on at least one of these aspects: competencies, relevant laws, articles references (missing or hallucination). 
+            Draft a complete explanation of your reasoning, mention the legal issue(s), the correct complete relevant articles used in both eur lex and national and the applicable regime set up, the conditions if applicable, conclusions."""
+        )
+    )
+
+    response = chat.invoke(main_wf_messages)
+    print('----------Complementary------------')
+    print(response.content)
+    export['ai_challange'] = str(response.content)
+    return export
+
+
+if __name__ == '__main__':
+    export = run_pipeline()
+    print(json.dumps(export, indent=4))
+    with open('export.json', 'w') as f:
+        json.dump(export, f, indent=4)

requirements.txt

@@ -0,0 +1,5 @@
+langchain
+langchain-community
+langchain-aws
+boto3
+gradio