Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make nft web app nicer #602

Merged
merged 2 commits into from
Dec 13, 2024
Merged

Make nft web app nicer #602

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cdaa11e
Make nft web app nicer
kongzii Dec 13, 2024
f58818d
little updates
kongzii Dec 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 50 additions & 36 deletions prediction_market_agent/agents/microchain_agent/nft_treasury_game/app_nft_treasury_game.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@
import streamlit as st
from microchain.functions import Reasoning
from prediction_market_agent_tooling.tools.balances import get_balances
from prediction_market_agent_tooling.tools.datetime_utc import DatetimeUTC
from prediction_market_agent_tooling.tools.utils import check_not_none
from streamlit_extras.stylable_container import stylable_container

from prediction_market_agent.agents.identifiers import AgentIdentifier
from prediction_market_agent.agents.microchain_agent.messages_functions import (
Expand All @@ -27,6 +28,7 @@
DeployableAgentNFTGameAbstract,
)
from prediction_market_agent.db.long_term_memory_table_handler import (
LongTermMemories,
LongTermMemoryTableHandler,
)
from prediction_market_agent.db.prompt_table_handler import PromptTableHandler
Expand Down Expand Up @@ -63,8 +65,8 @@ def send_message_part(nft_agent: type[DeployableAgentNFTGameAbstract]) -> None:


def parse_function_and_body(
role: t.Literal["user", "assistant", "system"], message: str
) -> t.Tuple[str | None, str | None]:
role: t.Literal["user", "assistant"], message: str
) -> t.Tuple[str, str]:
Comment on lines +68 to +69
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Codebase verification

Based on the gathered evidence, I can now provide a final response:

The "system" role removal requires attention in specific files

The codebase still contains several references to the "system" role that need to be addressed:

  • prediction_market_agent/agents/microchain_agent/agent_functions.py: Uses "system" role in history assignment
  • prediction_market_agent/agents/microchain_agent/microchain_agent.py: Contains two instances of "system" role usage
  • prediction_market_agent/agents/microchain_agent/memory.py: Has a comparison with "system" role
  • prediction_market_agent/agents/microchain_agent/nft_treasury_game/app_nft_treasury_game.py: Contains filtering logic for "system" role

While the parse_function_and_body function has been updated to exclude the "system" role, there are still active uses of it in the codebase that need to be reviewed and potentially updated for consistency.

🔗 Analysis chain

Confirm removal of the "system" role throughout the codebase

The parse_function_and_body function now excludes the "system" role. Please verify that all references to the "system" role have been updated elsewhere to prevent potential inconsistencies or errors.

Run the following script to check for any remaining references to the "system" role:

Let me gather more context about the usage of roles in the codebase, particularly around the parse_function_and_body function.

Let me try a different approach to understand the context better.

🏁 Scripts executed

The following scripts were executed for the analysis:

Script:

#!/bin/bash
# Description: Search for references to the "system" role in the codebase.

# Test: Find occurrences where "role" is assigned or compared to "system"
rg 'role.*["'\'']system["'\'']'

Length of output: 1616

Script:

#!/bin/bash
# Check the implementation of parse_function_and_body and surrounding context
ast-grep --pattern 'def parse_function_and_body($$$) {
  $$$
}'

# Also check for any type definitions or constants related to roles
rg -A 3 'Literal\[.*"system".*\]|ROLE_.*system'

Length of output: 115

Script:

#!/bin/bash
# Get the full content of the file containing parse_function_and_body
fd "app_nft_treasury_game.py" --exec cat {}

# Look for any role-related type definitions or enums
rg "Literal\[.*role.*\]|Role|ROLE"

Length of output: 10233

message = message.strip()

if role == "assistant":
Expand All @@ -75,37 +77,30 @@ def parse_function_and_body(
# Responses from the individual functions are stored under `user` role.
parsed_function = DummyFunctionName.RESPONSE_FUNCTION_NAME
parsed_body = message
elif role == "system":
# System message isn't shown in the chat history, so ignore.
parsed_function = None
parsed_body = None
else:
raise ValueError(f"Unknown role: {role}")

return parsed_function, parsed_body


def customized_chat_message(
role: t.Literal["user", "assistant", "system"],
message: str,
created_at: DatetimeUTC,
function_call: LongTermMemories,
function_output: LongTermMemories,
) -> None:
parsed_function, parsed_body = parse_function_and_body(role, message)
if parsed_function is None:
return
# If the message is output from one of these functions, skip it, because it's not interesting to read `The reasoning has been recorded` and similar over and over again.
if parsed_body in (
Reasoning()(""),
BroadcastPublicMessageToHumans.OUTPUT_TEXT,
SendPaidMessageToAnotherAgent.OUTPUT_TEXT,
):
return
created_at = function_output.datetime_

parsed_function_call_name, parsed_function_call_body = parse_function_and_body(
check_not_none(function_call.metadata_dict)["role"],
check_not_none(function_call.metadata_dict)["content"],
)
parsed_function_output_name, parsed_function_output_body = parse_function_and_body(
check_not_none(function_output.metadata_dict)["role"],
check_not_none(function_output.metadata_dict)["content"],
)
Comment on lines +87 to +99
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Handle potential KeyError when accessing metadata_dict keys

Accessing metadata_dict["role"] and metadata_dict["content"] without checking if the keys exist may lead to a KeyError. Consider using the .get() method with default values or adding error handling to ensure robustness.

Apply this diff to prevent potential KeyError exceptions:

 parsed_function_call_name, parsed_function_call_body = parse_function_and_body(
-    check_not_none(function_call.metadata_dict)["role"],
-    check_not_none(function_call.metadata_dict)["content"],
+    check_not_none(function_call.metadata_dict.get("role", "")),
+    check_not_none(function_call.metadata_dict.get("content", "")),
 )

 parsed_function_output_name, parsed_function_output_body = parse_function_and_body(
-    check_not_none(function_output.metadata_dict)["role"],
-    check_not_none(function_output.metadata_dict)["content"],
+    check_not_none(function_output.metadata_dict.get("role", "")),
+    check_not_none(function_output.metadata_dict.get("content", "")),
 )
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
function_call: LongTermMemories,
function_output: LongTermMemories,
) -> None:
parsed_function, parsed_body = parse_function_and_body(role, message)
if parsed_function is None:
return
# If the message is output from one of these functions, skip it, because it's not interesting to read `The reasoning has been recorded` and similar over and over again.
if parsed_body in (
Reasoning()(""),
BroadcastPublicMessageToHumans.OUTPUT_TEXT,
SendPaidMessageToAnotherAgent.OUTPUT_TEXT,
):
return
created_at = function_output.datetime_
parsed_function_call_name, parsed_function_call_body = parse_function_and_body(
check_not_none(function_call.metadata_dict)["role"],
check_not_none(function_call.metadata_dict)["content"],
)
parsed_function_output_name, parsed_function_output_body = parse_function_and_body(
check_not_none(function_output.metadata_dict)["role"],
check_not_none(function_output.metadata_dict)["content"],
)
function_call: LongTermMemories,
function_output: LongTermMemories,
) -> None:
created_at = function_output.datetime_
parsed_function_call_name, parsed_function_call_body = parse_function_and_body(
check_not_none(function_call.metadata_dict.get("role", "")),
check_not_none(function_call.metadata_dict.get("content", "")),
)
parsed_function_output_name, parsed_function_output_body = parse_function_and_body(
check_not_none(function_output.metadata_dict.get("role", "")),
check_not_none(function_output.metadata_dict.get("content", "")),
)


match parsed_function:
match parsed_function_call_name:
case Reasoning.__name__:
icon = "🧠"
case DummyFunctionName.RESPONSE_FUNCTION_NAME:
icon = "✔️"
case ReceiveMessage.__name__:
icon = "👤"
case BroadcastPublicMessageToHumans.__name__:
Expand All @@ -116,11 +111,24 @@ def customized_chat_message(
icon = "🤖"

with st.chat_message(icon):
if parsed_function:
st.markdown(f"**{parsed_function}**")
st.write(created_at.strftime("%Y-%m-%d %H:%M:%S"))
if message:
st.markdown(parsed_body)
if parsed_function_call_name == Reasoning.__name__:
# Don't show reasoning as function call, to make it a bit nicer.
st.markdown(
parsed_function_call_body.replace("reasoning='", "").replace("')", "")
)
else:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Improve parsing of parsed_function_call_body for robustness

Using string replacements to parse parsed_function_call_body may not handle all cases correctly. Consider using regular expressions or parsing the arguments properly to make the code more robust.

Apply this diff to use a regular expression for parsing:

+import re
...
     if parsed_function_call_name == Reasoning.__name__:
         # Don't show reasoning as function call, to make it a bit nicer.
-        st.markdown(
-            parsed_function_call_body.replace("reasoning='", "").replace("')", "")
-        )
+        reasoning_match = re.match(r"reasoning=['"](.*)['"]", parsed_function_call_body)
+        if reasoning_match:
+            reasoning_text = reasoning_match.group(1)
+            st.markdown(reasoning_text)
+        else:
+            st.markdown(parsed_function_call_body)

Committable suggestion skipped: line range outside the PR's diff.

# Otherwise, show it as a normal function-response call, e.g. `ReceiveMessages() -> ...`.
st.markdown(
f"**{parsed_function_call_name}**({parsed_function_call_body}) *{created_at.strftime('%Y-%m-%d %H:%M:%S')}*"
)

# Only show the output if it's supposed to be interesting.
if parsed_function_call_name not in (
Reasoning.__name__,
BroadcastPublicMessageToHumans.__name__,
SendPaidMessageToAnotherAgent.__name__,
):
st.markdown(parsed_function_output_body)


@st.fragment(run_every=timedelta(seconds=5))
Expand All @@ -134,14 +142,20 @@ def show_function_calls_part(nft_agent: type[DeployableAgentNFTGameAbstract]) ->
st.markdown("No actions yet.")
return

for item in calls:
if item.metadata_dict is None:
continue
customized_chat_message(
item.metadata_dict["role"],
item.metadata_dict["content"],
item.datetime_,
)
# Filter out system calls, because they aren't supposed to be shown in the chat history itself.
calls = [
call for call in calls if check_not_none(call.metadata_dict)["role"] != "system"
]

# Microchain works on `function call` - `funtion response` pairs, so we will process them together.
for index, (function_output, function_call) in enumerate(
zip(calls[::2], calls[1::2])
):
with stylable_container(
key=f"function_call_{index}",
css_styles=f"{{background-color: {'#f0f0f0' if (index % 2 == 0) else 'white'}; border-radius: 5px;}}",
):
customized_chat_message(function_call, function_output)


@st.fragment(run_every=timedelta(seconds=5))
Expand Down
2 changes: 1 addition & 1 deletion prediction_market_agent/db/long_term_memory_table_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ def search(
from_: DatetimeUTC | None = None,
to_: DatetimeUTC | None = None,
limit: int | None = None,
) -> t.Sequence[LongTermMemories]:
) -> list[LongTermMemories]:
"""Searches the LongTermMemoryTableHandler for entries within a specified datetime range that match
self.task_description."""
query_filters = [
Expand Down
2 changes: 1 addition & 1 deletion prediction_market_agent/db/sql_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ def get_with_filter_and_order(
order_by_column_name: str | None = None,
order_desc: bool = True,
limit: int | None = None,
) -> t.Sequence[SQLModelType]:
) -> list[SQLModelType]:
with self.db_manager.get_session() as session:
query = session.query(self.table)
for exp in query_filters:
Expand Down
Loading