Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/strategies/parsing.py: 13%

1"""Shared parsing utilities for agent strategies.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.ai.llm import ChatMessage, Role 

8 

9 

10def extract_thought(text: str) -> str: 

11 """Extract the THOUGHT section from the LLM response.""" 

12 for line in text.split("\n"): 

13 stripped = line.strip() 

14 if stripped.upper().startswith("THOUGHT:"): 

15 return stripped[len("THOUGHT:") :].strip() 

16 return text.split("\n", maxsplit=1)[0][:200] 

17 

18 

19def extract_final_answer(text: str) -> str | None: 

20 """Extract FINAL_ANSWER if present.""" 

21 marker = "FINAL_ANSWER:" 

22 upper = text.upper() 

23 idx = upper.find(marker) 

24 if idx == -1: 

25 return None 

26 return text[idx + len(marker) :].strip() 

27 

28 

29def extract_tool_call(text: str) -> tuple[str | None, dict[str, Any]]: 

30 """Extract ACTION and ACTION_INPUT from the LLM response.""" 

31 from lexigram.serialization.backends.json import JSONDecodeError, loads 

32 

33 action_name: str | None = None 

34 action_input: dict[str, Any] = {} 

35 

36 for line in text.split("\n"): 

37 stripped = line.strip() 

38 upper = stripped.upper() 

39 if upper.startswith("ACTION:") and not upper.startswith("ACTION_INPUT:"): 

40 action_name = stripped[len("ACTION:") :].strip() 

41 elif upper.startswith("ACTION_INPUT:"): 

42 raw = stripped[len("ACTION_INPUT:") :].strip() 

43 try: 

44 action_input = loads(raw) 

45 except (ValueError, JSONDecodeError): 

46 remaining = text[text.index(stripped) :] 

47 brace_start = remaining.find("{") 

48 if brace_start != -1: 

49 depth = 0 

50 for i, ch in enumerate(remaining[brace_start:]): 

51 if ch == "{": 

52 depth += 1 

53 elif ch == "}": 

54 depth -= 1 

55 if depth == 0: 

56 try: 

57 action_input = loads( 

58 remaining[brace_start : brace_start + i + 1] 

59 ) 

60 except (ValueError, JSONDecodeError): 

61 pass 

62 break 

63 

64 return action_name, action_input 

65 

66 

67def build_chat_messages( 

68 message: str, 

69 history: list[ChatMessage], 

70 system_prompt: str, 

71) -> list[ChatMessage]: 

72 """Convert history + new message into ChatMessage objects.""" 

73 messages: list[ChatMessage] = [] 

74 

75 if system_prompt: 

76 messages.append(ChatMessage(role=Role.SYSTEM, content=system_prompt)) 

77 

78 messages.extend(history) 

79 

80 messages.append(ChatMessage(role=Role.USER, content=message)) 

81 return messages 

82 

83 

84def build_chat_messages_from_dict( 

85 message: str, 

86 history: list[dict[str, Any]], 

87 system_prompt: str, 

88) -> list[ChatMessage]: 

89 """Convert history (as dicts) + new message into ChatMessage objects.""" 

90 messages: list[ChatMessage] = [] 

91 

92 if system_prompt: 

93 messages.append(ChatMessage(role=Role.SYSTEM, content=system_prompt)) 

94 

95 for entry in history: 

96 role_str = entry.get("role", "user") 

97 content = entry.get("content", "") 

98 try: 

99 role = Role(role_str) 

100 except ValueError: 

101 role = Role.USER 

102 messages.append(ChatMessage(role=role, content=content)) 

103 

104 messages.append(ChatMessage(role=Role.USER, content=message)) 

105 return messages 

106 

107 

108__all__ = [ 

109 "build_chat_messages", 

110 "build_chat_messages_from_dict", 

111 "extract_final_answer", 

112 "extract_thought", 

113 "extract_tool_call", 

114]