Spaces:
Sleeping
Sleeping
File size: 1,931 Bytes
5db30ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | from langgraph.prebuilt.chat_agent_executor import AgentState
from langchain_core.messages import BaseMessage
from typing import Union, Sequence, Callable, Tuple, Dict
from elbench.basic_education_runtime.utils import content_to_text, remove_think
def any_keyword_route(
keywords: Sequence[str], exists_to: str, else_to: str, think_as_message: bool = False
) -> Tuple[Callable[..., bool], Dict[bool, str]]:
"""Route based on keywords."""
path_map = {
True: exists_to,
False: else_to,
}
def route(state: Union[AgentState, Sequence[BaseMessage]]) -> bool:
"""Route based on keywords."""
if isinstance(state, Sequence):
message = state[-1]
elif messages := state.get("messages", []):
message = messages[-1]
else:
raise ValueError("No messages found, error while routing")
content = message.content
if not think_as_message:
content = remove_think(content) # Remove think tags
text = content_to_text(content, include_reasoning=think_as_message).lower()
return any(keyword in text for keyword in keywords)
return (route, path_map)
def all_keyword_route(
keywords: Sequence[str], exists_to: str, else_to: str
) -> Tuple[Callable[..., bool], Dict[bool, str]]:
"""Route based on keywords."""
path_map = {
True: exists_to,
False: else_to,
}
def route(state: Union[AgentState, Sequence[BaseMessage]]) -> bool:
"""Route based on keywords."""
if isinstance(state, Sequence):
message = state[-1]
elif messages := state.get("messages", []):
message = messages[-1]
else:
raise ValueError("No messages found, error while routing")
text = content_to_text(message.content).lower()
return all(keyword in text for keyword in keywords)
return (route, path_map)
|