Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/parsers/csv.py: 42%

1"""CSV Output Parser.""" 

2 

3from __future__ import annotations 

4 

5import csv 

6import io 

7from typing import Any 

8 

9from lexigram.ai.llm.structured.exceptions import ParseError 

10from lexigram.ai.llm.structured.parser import extract_json_block 

11from lexigram.logging import ( 

12 get_logger, 

13) 

14 

15logger = get_logger(__name__) 

16 

17 

18class CSVOutputParser: 

19 """Parse LLM responses into lists of dictionaries (CSV format). 

20 

21 Extracts JSON array from the response and converts to list of dicts, 

22 where each dict represents a CSV row with column names as keys. 

23 

24 Example: 

25 >>> parser = CSVOutputParser() 

26 >>> result = parser.parse('[{"name": "John", "age": 30}, {"name": "Jane", "age": 25}]') 

27 >>> assert len(result) == 2 

28 >>> assert result[0]["name"] == "John" 

29 """ 

30 

31 def parse(self, text: str) -> list[dict[str, Any]]: 

32 """Parse text into a list of dictionaries. 

33 

34 Args: 

35 text: Raw LLM response text that may contain JSON array. 

36 

37 Returns: 

38 List of dictionaries, each representing a CSV row. 

39 

40 Raises: 

41 ParseError: When JSON cannot be extracted or is not an array. 

42 """ 

43 try: 

44 parsed = extract_json_block(text) 

45 except ValueError as exc: 

46 raise ParseError(str(exc)) from exc 

47 

48 if not isinstance(parsed, list): 

49 raise ParseError( 

50 f"Expected JSON array for CSV, got {type(parsed).__name__}" 

51 ) 

52 

53 for i, item in enumerate(parsed): 

54 if not isinstance(item, dict): 

55 raise ParseError(f"Row {i} is not a dict, got {type(item).__name__}") 

56 

57 logger.debug("csv_parsed", row_count=len(parsed)) 

58 return parsed 

59 

60 def parse_csv_string(self, text: str) -> list[dict[str, Any]]: 

61 """Parse raw CSV text (not JSON) into list of dictionaries. 

62 

63 Args: 

64 text: Raw CSV text with header row. 

65 

66 Returns: 

67 List of dictionaries, each representing a CSV row. 

68 

69 Raises: 

70 ParseError: When CSV cannot be parsed. 

71 """ 

72 text = text.strip() 

73 try: 

74 reader = csv.DictReader(io.StringIO(text)) 

75 return list(reader) 

76 except csv.Error as exc: 

77 raise ParseError(f"Failed to parse CSV: {exc}") from exc 

78 

79 def get_format_instructions(self) -> str: 

80 """Return format instructions for the LLM. 

81 

82 Returns: 

83 Format instruction string telling the model to output a valid 

84 JSON array of objects. 

85 """ 

86 return ( 

87 "Your response should be a valid JSON array of objects. " 

88 "Each object represents a row with column names as keys. " 

89 "Do not include any text before or after the JSON array. " 

90 "Do not use markdown code fences." 

91 ) 

92 

93 

94__all__ = ["CSVOutputParser"]