|
|
from typing import Tuple |
|
|
|
|
|
import logging |
|
|
|
|
|
from langchain_core.messages import AIMessage, HumanMessage |
|
|
|
|
|
def add_details(response: str, reasoning: str, svg_argmap: str) -> str: |
|
|
"""Add reasoning details to the response message shown in chat.""" |
|
|
response_with_details = ( |
|
|
f"<p>{response}</p>" |
|
|
'<details id="reasoning">' |
|
|
"<summary><i>Internal reasoning trace</i></summary>" |
|
|
f"<code>{reasoning}</code></details>" |
|
|
'<details id="svg_argmap">' |
|
|
"<summary><i>Argument map</i></summary>" |
|
|
f"\n<div>\n{svg_argmap}\n</div>\n</details>" |
|
|
) |
|
|
return response_with_details |
|
|
|
|
|
|
|
|
def get_details(response_with_details: str) -> Tuple[str, dict[str, str]]: |
|
|
"""Extract response and details from response_with_details shown in chat.""" |
|
|
if "<details id=" not in response_with_details: |
|
|
return response_with_details, {} |
|
|
details_dict = {} |
|
|
response, *details_raw = response_with_details.split('<details id="') |
|
|
for details in details_raw: |
|
|
details_id, details_content = details.split('"', maxsplit=1) |
|
|
details_content = details_content.strip() |
|
|
if details_content.endswith("</code></details>"): |
|
|
details_content = details_content.split("<code>")[1].strip() |
|
|
details_content = details_content[:-len("</code></details>")].strip() |
|
|
elif details_content.endswith("</div></details>"): |
|
|
details_content = details_content.split("<div>")[1].strip() |
|
|
details_content = details_content[:-len("</div></details>")].strip() |
|
|
else: |
|
|
logging.warning(f"Unrecognized details content: {details_content}") |
|
|
details_content = "UNRECOGNIZED DETAILS CONTENT" |
|
|
details_dict[details_id] = details_content |
|
|
return response, details_dict |
|
|
|
|
|
|
|
|
def history_to_langchain_format(history: list[tuple[str, str]]) -> list: |
|
|
history_langchain_format = [] |
|
|
for human, ai in history: |
|
|
history_langchain_format.append(HumanMessage(content=human)) |
|
|
if ai is None: |
|
|
continue |
|
|
response, details = get_details(ai) |
|
|
logging.debug(f"Details: {details}") |
|
|
content = response |
|
|
if "reasoning" in details: |
|
|
content += ( |
|
|
"\n\n" |
|
|
"#+BEGIN_INTERNAL_TRACE // Internal reasoning trace (hidden from user)\n" |
|
|
f"{details.get('reasoning', '')}\n" |
|
|
"#+END_INTERNAL_TRACE" |
|
|
) |
|
|
history_langchain_format.append(AIMessage(content=content)) |
|
|
return history_langchain_format |
|
|
|
