Zum Inhalt

API Referenz

Diese Dokumentation wird automatisch aus den Docstrings des Quellcodes generiert.

Core Services

RecognitionService

modul_anerkennung.services.RecognitionService

Service layer for module recognition logic.

Source code in modul_anerkennung/services.py 
 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
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class RecognitionService:
    """Service layer for module recognition logic."""

    def __init__(self, llm: LLMInterface = None):
        """Initialisiert den RecognitionService.

        Args:
            llm (LLMInterface, optional): Die Schnittstelle zum LLM. Falls None, wird
                eine neue Instanz von LLMInterface erstellt.
        """
        self.llm = llm or LLMInterface()

    async def analyze_module(self, text: str) -> ModuleAnalysis:
        """Analyzes an external module description to extract name, ECTS, and keywords."""
        if not text:
            raise ValueError("Keine Modulbeschreibung angegeben.")

        logger.debug(f"Analysiere Modultext (Länge: {len(text)})")
        prompt = f"""Analysiere die folgende Modulbeschreibung und extrahiere:
1. Modulname
2. Anzahl ECTS (nur die Zahl)
3. 3-4 prägnante Suchbegriffe für eine semantische Suche.

Antworte ausschließlich im JSON-Format:
{{
  "name": "...",
  "ects": 5,
  "keywords": ["...", "...", "..."]
}}

Modulbeschreibung:
{text}"""

        response = await self.llm.achat([{"role": "user", "content": prompt}])
        analysis = self.llm.extract_json(response, ModuleAnalysis)
        logger.debug(f"Extrahiertes Modul: {analysis.name} ({analysis.ects} ECTS)")
        return analysis

    async def search_and_compare(
        self, po_id: str, keywords: str, max_ects: str, external_text: str
    ) -> List[Tuple[Dict[str, Any], ComparisonReport]]:
        """Searches for similar internal modules and compares them to the external module."""
        if not po_id:
            logger.warning("Keine PO-ID angegeben für die Suche.")
            return []

        try:
            ects_val = float(max_ects) if max_ects else None
        except ValueError:
            ects_val = None

        logger.debug(f"Suche nach Modulen für PO {po_id} mit Keywords: {keywords}")
        async with MocogiClient() as client:
            modules = await client.call_tool(
                "search_modules",
                {"po_id": po_id, "search_term": keywords, "max_ects": ects_val},
            )

        logger.debug(f"Gefundene Module: {len(modules)}")
        # Process top 5 modules
        results = []
        for m in modules[:5]:
            comp = await self.perform_comparison(external_text, m)
            results.append((m, comp))

        return results

    async def perform_comparison(
        self, external_text: str, internal_module: Dict[str, Any]
    ) -> ComparisonReport:
        """Performs a detailed comparison between an external and an internal module."""
        m_title = internal_module.get("metadata", {}).get("title", "Unbekannt")
        logger.debug(f"Vergleiche mit internem Modul: {m_title}")

        internal_text = json.dumps(internal_module, indent=2)

        prompt = f"""Vergleiche die folgende externe Modulbeschreibung mit unserem internen Modul.

Externe Beschreibung:
{external_text}

Internes Modul:
{internal_text}

Erstelle einen detaillierten Vergleichsbericht.
Bestimme, ob das Modul anerkannt werden kann (Ja, Nein, Vielleicht).
Antworte im JSON-Format:
{{
  "decision": "Ja" | "Nein" | "Vielleicht",
  "reasoning": "Kurze Begründung",
  "report": "Ausführlicher Bericht"
}}
"""
        response = await self.llm.achat([{"role": "user", "content": prompt}])
        comparison = self.llm.extract_json(response, ComparisonReport)
        logger.debug(f"Ergebnis für {m_title}: {comparison.decision}")
        return comparison

Functions

__init__(llm=None)

Initialisiert den RecognitionService.

Parameters:

Name Type Description Default
llm LLMInterface

Die Schnittstelle zum LLM. Falls None, wird eine neue Instanz von LLMInterface erstellt.

 None
Source code in modul_anerkennung/services.py 
16
17
18
19
20
21
22
23
def __init__(self, llm: LLMInterface = None):
    """Initialisiert den RecognitionService.

    Args:
        llm (LLMInterface, optional): Die Schnittstelle zum LLM. Falls None, wird
            eine neue Instanz von LLMInterface erstellt.
    """
    self.llm = llm or LLMInterface()

analyze_module(text) async

Analyzes an external module description to extract name, ECTS, and keywords.

Source code in modul_anerkennung/services.py 
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
    async def analyze_module(self, text: str) -> ModuleAnalysis:
        """Analyzes an external module description to extract name, ECTS, and keywords."""
        if not text:
            raise ValueError("Keine Modulbeschreibung angegeben.")

        logger.debug(f"Analysiere Modultext (Länge: {len(text)})")
        prompt = f"""Analysiere die folgende Modulbeschreibung und extrahiere:
1. Modulname
2. Anzahl ECTS (nur die Zahl)
3. 3-4 prägnante Suchbegriffe für eine semantische Suche.

Antworte ausschließlich im JSON-Format:
{{
  "name": "...",
  "ects": 5,
  "keywords": ["...", "...", "..."]
}}

Modulbeschreibung:
{text}"""

        response = await self.llm.achat([{"role": "user", "content": prompt}])
        analysis = self.llm.extract_json(response, ModuleAnalysis)
        logger.debug(f"Extrahiertes Modul: {analysis.name} ({analysis.ects} ECTS)")
        return analysis

perform_comparison(external_text, internal_module) async

Performs a detailed comparison between an external and an internal module.

Source code in modul_anerkennung/services.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
    async def perform_comparison(
        self, external_text: str, internal_module: Dict[str, Any]
    ) -> ComparisonReport:
        """Performs a detailed comparison between an external and an internal module."""
        m_title = internal_module.get("metadata", {}).get("title", "Unbekannt")
        logger.debug(f"Vergleiche mit internem Modul: {m_title}")

        internal_text = json.dumps(internal_module, indent=2)

        prompt = f"""Vergleiche die folgende externe Modulbeschreibung mit unserem internen Modul.

Externe Beschreibung:
{external_text}

Internes Modul:
{internal_text}

Erstelle einen detaillierten Vergleichsbericht.
Bestimme, ob das Modul anerkannt werden kann (Ja, Nein, Vielleicht).
Antworte im JSON-Format:
{{
  "decision": "Ja" | "Nein" | "Vielleicht",
  "reasoning": "Kurze Begründung",
  "report": "Ausführlicher Bericht"
}}
"""
        response = await self.llm.achat([{"role": "user", "content": prompt}])
        comparison = self.llm.extract_json(response, ComparisonReport)
        logger.debug(f"Ergebnis für {m_title}: {comparison.decision}")
        return comparison

search_and_compare(po_id, keywords, max_ects, external_text) async

Searches for similar internal modules and compares them to the external module.

Source code in modul_anerkennung/services.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def search_and_compare(
    self, po_id: str, keywords: str, max_ects: str, external_text: str
) -> List[Tuple[Dict[str, Any], ComparisonReport]]:
    """Searches for similar internal modules and compares them to the external module."""
    if not po_id:
        logger.warning("Keine PO-ID angegeben für die Suche.")
        return []

    try:
        ects_val = float(max_ects) if max_ects else None
    except ValueError:
        ects_val = None

    logger.debug(f"Suche nach Modulen für PO {po_id} mit Keywords: {keywords}")
    async with MocogiClient() as client:
        modules = await client.call_tool(
            "search_modules",
            {"po_id": po_id, "search_term": keywords, "max_ects": ects_val},
        )

    logger.debug(f"Gefundene Module: {len(modules)}")
    # Process top 5 modules
    results = []
    for m in modules[:5]:
        comp = await self.perform_comparison(external_text, m)
        results.append((m, comp))

    return results

MCP Integration

MocogiClient

modul_anerkennung.mcp_client.MocogiClient

Ein Client für den Mocogi MCP Server, der die Kommunikation über stdio ermöglicht.

Dieser Client startet den Mocogi MCP Server als Subprozess und ermöglicht den Zugriff auf dessen Tools über eine asynchrone Schnittstelle.

Source code in modul_anerkennung/mcp_client.py 
 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
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
class MocogiClient:
    """Ein Client für den Mocogi MCP Server, der die Kommunikation über stdio ermöglicht.

    Dieser Client startet den Mocogi MCP Server als Subprozess und ermöglicht
    den Zugriff auf dessen Tools über eine asynchrone Schnittstelle.
    """

    def __init__(self):
        """Initialisiert den MocogiClient und konfiguriert den Server-Befehl.

        Falls fileno nicht unterstützt wird (z.B. in Jupyter/Colab), wird auf
        einen In-Memory-Client ausgewichen.
        """
        use_stdio = True
        try:
            sys.stdin.fileno()
            sys.stdout.fileno()
        except (io.UnsupportedOperation, AttributeError):
            use_stdio = False

        if use_stdio:
            logger.info("Initialisiere MocogiClient mit StdioTransport")
            # Der Server wird als Subprozess gestartet
            transport = StdioTransport(
                command=sys.executable, args=["-m", "modul_anerkennung.mocogi_mcp"]
            )
            self.client = Client(transport)
        else:
            logger.info("Initialisiere MocogiClient mit In-Memory-Client (Fallback)")
            # Fallback: In-Memory Client für Umgebungen ohne fileno (Jupyter/Colab)
            try:
                from modul_anerkennung.mocogi_mcp import mcp as server

                self.client = Client(server)
            except ImportError:
                # Falls Import fehlschlägt (z.B. bei Installation als Package)
                # versuchen wir es mit dem Namen
                self.client = Client("modul_anerkennung.mocogi_mcp")

    async def __aenter__(self):
        """Ermöglicht die Nutzung des Clients als asynchroner Kontextmanager.

        Returns:
            MocogiClient: Die Instanz des Clients.
        """
        await self.client.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Beendet den Client beim Verlassen des asynchronen Kontextmanagers.

        Args:
            exc_type: Der Typ der Ausnahme, falls eine aufgetreten ist.
            exc_val: Der Wert der Ausnahme, falls eine aufgetreten ist.
            exc_tb: Der Traceback der Ausnahme, falls eine aufgetreten ist.
        """
        await self.client.__aexit__(exc_type, exc_val, exc_tb)

    async def list_study_programs(
        self, filter: str = "currently-active"
    ) -> List[Dict[str, Any]]:
        """Listet alle Studiengänge der TH Köln auf.

        Args:
            filter (str, optional): Filter für die Studiengänge (z.B. 'currently-active').
                Standardwert ist "currently-active".

        Returns:
            List[Dict[str, Any]]: Eine Liste von Studiengängen als Dictionaries.
        """
        return await self.call_tool("list_study_programs", {"filter": filter})

    async def get_modules_by_po(self, po_id: str) -> List[Dict[str, Any]]:
        """Gibt alle aktiven Module für eine bestimmte Prüfungsordnung (PO) zurück.

        Args:
            po_id (str): Die ID der Prüfungsordnung (z.B. 'inf_mi5').

        Returns:
            List[Dict[str, Any]]: Eine Liste von Modulen als Dictionaries.
        """
        return await self.call_tool("get_modules_by_po", {"po_id": po_id})

    async def get_all_active_modules(self) -> List[Dict[str, Any]]:
        """Gibt eine Liste aller aktiven Module zurück.

        Returns:
            List[Dict[str, Any]]: Eine Liste aller aktiven Module.
        """
        return await self.call_tool("get_all_active_modules")

    async def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Any:
        """Führt einen generischen Aufruf eines MCP-Tools aus.

        Args:
            name (str): Der Name des aufzurufenden Tools.
            arguments (Dict[str, Any], optional): Die Argumente für das Tool.
                Standardwert ist None.

        Returns:
            Any: Das Ergebnis des Tool-Aufrufs.
        """
        logger.debug(f"Rufe Tool auf: {name} mit Argumenten: {arguments}")
        result = await self.client.call_tool(name, arguments or {})

        # FastMCP 3.x liefert ein CallToolResult Objekt zurück.
        # Wir extrahieren die Daten für die JSON-Serialisierung.
        if hasattr(result, "data"):
            return result.data
        return result

    async def list_tools(self) -> List[Any]:
        """Listet alle verfügbaren Tools des MCP Servers auf.

        Returns:
            List[Any]: Eine Liste der verfügbaren Tools.
        """
        return await self.client.list_tools()

    async def get_tools_for_llm(self) -> List[Dict[str, Any]]:
        """Konvertiert MCP Tools in das Format für llm_client (OpenAI/Gemini Format).

        Returns:
            List[Dict[str, Any]]: Eine Liste von Tools im OpenAI-Format.
        """
        mcp_tools = await self.list_tools()
        tools = []
        for tool in mcp_tools:
            tools.append(
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema,
                    },
                }
            )
        return tools

    async def chat_with_tools(
        self, llm_client: Any, messages: List[Dict[str, Any]], max_iterations: int = 5
    ) -> str:
        """Führt einen Chat mit Tool-Unterstützung durch.

        Args:
            llm_client (Any): Der LLMClient (aus llm_client Package).
            messages (List[Dict[str, Any]]): Der bisherige Chat-Verlauf.
            max_iterations (int, optional): Maximale Anzahl an Tool-Call-Iterationen.
                Standardwert ist 5.

        Returns:
            str: Die Antwort des LLM.
        """
        tools = await self.get_tools_for_llm()

        for i in range(max_iterations):
            logger.info(f"Chat-Iteration {i + 1}/{max_iterations}")
            response = await llm_client.achat_completion_with_tools(
                messages=messages, tools=tools
            )

            # Falls das LLM direkt antwortet ohne Tool-Call
            if not response.get("tool_calls"):
                logger.info("LLM hat direkt geantwortet.")
                return response.get("content", "")

            # Falls Tool-Calls vorhanden sind, führen wir sie aus
            logger.info(f"LLM fordert {len(response['tool_calls'])} Tool-Calls an.")

            # WICHTIG: Den gesamten Response kopieren und als Assistant-Nachricht hinzufügen.
            # Dies stellt sicher, dass Felder wie 'thought' oder 'thought_signature' (Gemini)
            # erhalten bleiben, was für nachfolgende API-Aufrufe erforderlich ist.
            assistant_msg = response.copy()
            assistant_msg["role"] = "assistant"
            messages.append(assistant_msg)

            for tool_call in response["tool_calls"]:
                tool_name = tool_call["function"]["name"]
                tool_args = json.loads(tool_call["function"]["arguments"])

                # Tool über den MCP Client aufrufen
                logger.info(f"Führe Tool aus: {tool_name}")
                result = await self.call_tool(tool_name, tool_args)

                # Ergebnis zurück an den Chat-Verlauf übergeben
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "name": tool_name,
                        "content": json.dumps(result),
                    }
                )

        logger.warning("Maximale Anzahl an Tool-Calls erreicht.")
        return "Maximale Anzahl an Tool-Calls erreicht."

Functions

__aenter__() async

Ermöglicht die Nutzung des Clients als asynchroner Kontextmanager.

Returns:

Name Type Description
MocogiClient

Die Instanz des Clients.
Source code in modul_anerkennung/mcp_client.py 
54
55
56
57
58
59
60
61
async def __aenter__(self):
    """Ermöglicht die Nutzung des Clients als asynchroner Kontextmanager.

    Returns:
        MocogiClient: Die Instanz des Clients.
    """
    await self.client.__aenter__()
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Beendet den Client beim Verlassen des asynchronen Kontextmanagers.

Parameters:

Name Type Description Default
exc_type

Der Typ der Ausnahme, falls eine aufgetreten ist.

 required
exc_val

Der Wert der Ausnahme, falls eine aufgetreten ist.

 required
exc_tb

Der Traceback der Ausnahme, falls eine aufgetreten ist.

 required
Source code in modul_anerkennung/mcp_client.py 
63
64
65
66
67
68
69
70
71
async def __aexit__(self, exc_type, exc_val, exc_tb):
    """Beendet den Client beim Verlassen des asynchronen Kontextmanagers.

    Args:
        exc_type: Der Typ der Ausnahme, falls eine aufgetreten ist.
        exc_val: Der Wert der Ausnahme, falls eine aufgetreten ist.
        exc_tb: Der Traceback der Ausnahme, falls eine aufgetreten ist.
    """
    await self.client.__aexit__(exc_type, exc_val, exc_tb)

__init__()

Initialisiert den MocogiClient und konfiguriert den Server-Befehl.

Falls fileno nicht unterstützt wird (z.B. in Jupyter/Colab), wird auf einen In-Memory-Client ausgewichen.

Source code in modul_anerkennung/mcp_client.py 
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
def __init__(self):
    """Initialisiert den MocogiClient und konfiguriert den Server-Befehl.

    Falls fileno nicht unterstützt wird (z.B. in Jupyter/Colab), wird auf
    einen In-Memory-Client ausgewichen.
    """
    use_stdio = True
    try:
        sys.stdin.fileno()
        sys.stdout.fileno()
    except (io.UnsupportedOperation, AttributeError):
        use_stdio = False

    if use_stdio:
        logger.info("Initialisiere MocogiClient mit StdioTransport")
        # Der Server wird als Subprozess gestartet
        transport = StdioTransport(
            command=sys.executable, args=["-m", "modul_anerkennung.mocogi_mcp"]
        )
        self.client = Client(transport)
    else:
        logger.info("Initialisiere MocogiClient mit In-Memory-Client (Fallback)")
        # Fallback: In-Memory Client für Umgebungen ohne fileno (Jupyter/Colab)
        try:
            from modul_anerkennung.mocogi_mcp import mcp as server

            self.client = Client(server)
        except ImportError:
            # Falls Import fehlschlägt (z.B. bei Installation als Package)
            # versuchen wir es mit dem Namen
            self.client = Client("modul_anerkennung.mocogi_mcp")

call_tool(name, arguments=None) async

Führt einen generischen Aufruf eines MCP-Tools aus.

Parameters:

Name Type Description Default
name str

Der Name des aufzurufenden Tools.

 required
arguments Dict[str, Any]

Die Argumente für das Tool. Standardwert ist None.

 None

Returns:

Name Type Description
Any Any

Das Ergebnis des Tool-Aufrufs.
Source code in modul_anerkennung/mcp_client.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def call_tool(self, name: str, arguments: Dict[str, Any] = None) -> Any:
    """Führt einen generischen Aufruf eines MCP-Tools aus.

    Args:
        name (str): Der Name des aufzurufenden Tools.
        arguments (Dict[str, Any], optional): Die Argumente für das Tool.
            Standardwert ist None.

    Returns:
        Any: Das Ergebnis des Tool-Aufrufs.
    """
    logger.debug(f"Rufe Tool auf: {name} mit Argumenten: {arguments}")
    result = await self.client.call_tool(name, arguments or {})

    # FastMCP 3.x liefert ein CallToolResult Objekt zurück.
    # Wir extrahieren die Daten für die JSON-Serialisierung.
    if hasattr(result, "data"):
        return result.data
    return result

chat_with_tools(llm_client, messages, max_iterations=5) async

Führt einen Chat mit Tool-Unterstützung durch.

Parameters:

Name Type Description Default
llm_client Any

Der LLMClient (aus llm_client Package).

 required
messages List[Dict[str, Any]]

Der bisherige Chat-Verlauf.

 required
max_iterations int

Maximale Anzahl an Tool-Call-Iterationen. Standardwert ist 5.

 5

Returns:

Name Type Description
str str

Die Antwort des LLM.
Source code in modul_anerkennung/mcp_client.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def chat_with_tools(
    self, llm_client: Any, messages: List[Dict[str, Any]], max_iterations: int = 5
) -> str:
    """Führt einen Chat mit Tool-Unterstützung durch.

    Args:
        llm_client (Any): Der LLMClient (aus llm_client Package).
        messages (List[Dict[str, Any]]): Der bisherige Chat-Verlauf.
        max_iterations (int, optional): Maximale Anzahl an Tool-Call-Iterationen.
            Standardwert ist 5.

    Returns:
        str: Die Antwort des LLM.
    """
    tools = await self.get_tools_for_llm()

    for i in range(max_iterations):
        logger.info(f"Chat-Iteration {i + 1}/{max_iterations}")
        response = await llm_client.achat_completion_with_tools(
            messages=messages, tools=tools
        )

        # Falls das LLM direkt antwortet ohne Tool-Call
        if not response.get("tool_calls"):
            logger.info("LLM hat direkt geantwortet.")
            return response.get("content", "")

        # Falls Tool-Calls vorhanden sind, führen wir sie aus
        logger.info(f"LLM fordert {len(response['tool_calls'])} Tool-Calls an.")

        # WICHTIG: Den gesamten Response kopieren und als Assistant-Nachricht hinzufügen.
        # Dies stellt sicher, dass Felder wie 'thought' oder 'thought_signature' (Gemini)
        # erhalten bleiben, was für nachfolgende API-Aufrufe erforderlich ist.
        assistant_msg = response.copy()
        assistant_msg["role"] = "assistant"
        messages.append(assistant_msg)

        for tool_call in response["tool_calls"]:
            tool_name = tool_call["function"]["name"]
            tool_args = json.loads(tool_call["function"]["arguments"])

            # Tool über den MCP Client aufrufen
            logger.info(f"Führe Tool aus: {tool_name}")
            result = await self.call_tool(tool_name, tool_args)

            # Ergebnis zurück an den Chat-Verlauf übergeben
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "name": tool_name,
                    "content": json.dumps(result),
                }
            )

    logger.warning("Maximale Anzahl an Tool-Calls erreicht.")
    return "Maximale Anzahl an Tool-Calls erreicht."

get_all_active_modules() async

Gibt eine Liste aller aktiven Module zurück.

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Eine Liste aller aktiven Module.
Source code in modul_anerkennung/mcp_client.py 
 98
 99
100
101
102
103
104
async def get_all_active_modules(self) -> List[Dict[str, Any]]:
    """Gibt eine Liste aller aktiven Module zurück.

    Returns:
        List[Dict[str, Any]]: Eine Liste aller aktiven Module.
    """
    return await self.call_tool("get_all_active_modules")

get_modules_by_po(po_id) async

Gibt alle aktiven Module für eine bestimmte Prüfungsordnung (PO) zurück.

Parameters:

Name Type Description Default
po_id str

Die ID der Prüfungsordnung (z.B. 'inf_mi5').

 required

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Eine Liste von Modulen als Dictionaries.
Source code in modul_anerkennung/mcp_client.py 
87
88
89
90
91
92
93
94
95
96
async def get_modules_by_po(self, po_id: str) -> List[Dict[str, Any]]:
    """Gibt alle aktiven Module für eine bestimmte Prüfungsordnung (PO) zurück.

    Args:
        po_id (str): Die ID der Prüfungsordnung (z.B. 'inf_mi5').

    Returns:
        List[Dict[str, Any]]: Eine Liste von Modulen als Dictionaries.
    """
    return await self.call_tool("get_modules_by_po", {"po_id": po_id})

get_tools_for_llm() async

Konvertiert MCP Tools in das Format für llm_client (OpenAI/Gemini Format).

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Eine Liste von Tools im OpenAI-Format.
Source code in modul_anerkennung/mcp_client.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def get_tools_for_llm(self) -> List[Dict[str, Any]]:
    """Konvertiert MCP Tools in das Format für llm_client (OpenAI/Gemini Format).

    Returns:
        List[Dict[str, Any]]: Eine Liste von Tools im OpenAI-Format.
    """
    mcp_tools = await self.list_tools()
    tools = []
    for tool in mcp_tools:
        tools.append(
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                },
            }
        )
    return tools

list_study_programs(filter='currently-active') async

Listet alle Studiengänge der TH Köln auf.

Parameters:

Name Type Description Default
filter str

Filter für die Studiengänge (z.B. 'currently-active'). Standardwert ist "currently-active".

 'currently-active'

Returns:

Type Description
List[Dict[str, Any]]

List[Dict[str, Any]]: Eine Liste von Studiengängen als Dictionaries.
Source code in modul_anerkennung/mcp_client.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
async def list_study_programs(
    self, filter: str = "currently-active"
) -> List[Dict[str, Any]]:
    """Listet alle Studiengänge der TH Köln auf.

    Args:
        filter (str, optional): Filter für die Studiengänge (z.B. 'currently-active').
            Standardwert ist "currently-active".

    Returns:
        List[Dict[str, Any]]: Eine Liste von Studiengängen als Dictionaries.
    """
    return await self.call_tool("list_study_programs", {"filter": filter})

list_tools() async

Listet alle verfügbaren Tools des MCP Servers auf.

Returns:

Type Description
List[Any]

List[Any]: Eine Liste der verfügbaren Tools.
Source code in modul_anerkennung/mcp_client.py 
126
127
128
129
130
131
132
async def list_tools(self) -> List[Any]:
    """Listet alle verfügbaren Tools des MCP Servers auf.

    Returns:
        List[Any]: Eine Liste der verfügbaren Tools.
    """
    return await self.client.list_tools()

Datenmodelle

ModuleAnalysis

modul_anerkennung.models.ModuleAnalysis

Bases: BaseModel

Result of an LLM analysis of an external module description.

Source code in modul_anerkennung/models.py 
 7
 8
 9
10
11
12
class ModuleAnalysis(BaseModel):
    """Result of an LLM analysis of an external module description."""

    name: str = Field(..., description="Name of the external module")
    ects: Optional[float] = Field(None, description="Number of ECTS credits")
    keywords: List[str] = Field(default_factory=list, description="Keywords for search")

ComparisonReport

modul_anerkennung.models.ComparisonReport

Bases: BaseModel

Result of an LLM comparison between an external and internal module.

Source code in modul_anerkennung/models.py 
23
24
25
26
27
28
class ComparisonReport(BaseModel):
    """Result of an LLM comparison between an external and internal module."""

    decision: str = Field(..., description="Decision: Ja, Nein, or Vielleicht")
    reasoning: str = Field(..., description="Brief reasoning for the decision")
    report: str = Field(..., description="Detailed comparison report")

Hilfsklassen

LLMInterface

modul_anerkennung.llm_interface.LLMInterface

Wrapper für den universellen LLM-Client.

Source code in modul_anerkennung/llm_interface.py 
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class LLMInterface:
    """Wrapper für den universellen LLM-Client."""

    def __init__(self, provider: str | None = None, model: str | None = None) -> None:
        """
        Initialisiert den LLM-Client.

        Args:
            provider (str, optional): Der zu verwendende Provider (z.B. "openai", "groq", "gemini").
                                    Falls None, wird versucht, den Provider automatisch zu erkennen.
            model (str, optional): Das zu verwendende Modell.
        """
        # LLMClient lädt automatisch Keys aus der Umgebung
        self.client = LLMClient(api_choice=provider, llm=model)

    def chat(self, messages: List[Dict[str, str]], **kwargs: Any) -> str:
        """
        Führt eine Chat-Completion mit dem LLM aus.

        Args:
            messages (List[Dict[str, str]]): Nachrichtenverlauf für die LLM-Kommunikation.
            **kwargs: Zusätzliche Argumente for die Completion.
        Returns:
            str: Antwort des LLM.
        """
        return self.client.chat_completion(messages, **kwargs)

    async def achat(self, messages: List[Dict[str, str]], **kwargs: Any) -> str:
        """
        Führt eine asynchrone Chat-Completion mit dem LLM aus.

        Args:
            messages (List[Dict[str, str]]): Nachrichtenverlauf für die LLM-Kommunikation.
            **kwargs: Zusätzliche Argumente für die Completion.
        Returns:
            str: Antwort des LLM.
        """
        if not self.client.use_async:
            self.client.use_async = True
        return await self.client.achat_completion(messages, **kwargs)

    def extract_json(self, text: str, model_class: Type[T]) -> T:
        """
        Extrahiert JSON aus einem Text und validiert es gegen eine Pydantic-Klasse.

        Args:
            text (str): Der Text, der JSON enthält.
            model_class (Type[T]): Die Pydantic-Klasse zur Validierung.

        Returns:
            T: Die validierte Instanz der model_class.
        """
        # Suche nach dem ersten { und dem letzten }
        match = re.search(r"\{.*\}", text, re.DOTALL)
        if not match:
            raise ValueError(f"Kein JSON im LLM-Output gefunden: {text}")

        json_str = match.group(0)
        try:
            data = json.loads(json_str)
            return model_class.model_validate(data)
        except (json.JSONDecodeError, Exception) as e:
            raise ValueError(
                f"Fehler beim Parsen oder Validieren des JSON: {e}\nRaw JSON: {json_str}"
            )

Functions

__init__(provider=None, model=None)

Initialisiert den LLM-Client.

Parameters:

Name Type Description Default
provider str

Der zu verwendende Provider (z.B. "openai", "groq", "gemini"). Falls None, wird versucht, den Provider automatisch zu erkennen.

 None
model str

Das zu verwendende Modell.

 None
Source code in modul_anerkennung/llm_interface.py 
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, provider: str | None = None, model: str | None = None) -> None:
    """
    Initialisiert den LLM-Client.

    Args:
        provider (str, optional): Der zu verwendende Provider (z.B. "openai", "groq", "gemini").
                                Falls None, wird versucht, den Provider automatisch zu erkennen.
        model (str, optional): Das zu verwendende Modell.
    """
    # LLMClient lädt automatisch Keys aus der Umgebung
    self.client = LLMClient(api_choice=provider, llm=model)

achat(messages, **kwargs) async

Führt eine asynchrone Chat-Completion mit dem LLM aus.

Parameters:

Name Type Description Default
messages List[Dict[str, str]]

Nachrichtenverlauf für die LLM-Kommunikation.

 required
**kwargs Any

Zusätzliche Argumente für die Completion.

 {}

Returns: str: Antwort des LLM.

Source code in modul_anerkennung/llm_interface.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
async def achat(self, messages: List[Dict[str, str]], **kwargs: Any) -> str:
    """
    Führt eine asynchrone Chat-Completion mit dem LLM aus.

    Args:
        messages (List[Dict[str, str]]): Nachrichtenverlauf für die LLM-Kommunikation.
        **kwargs: Zusätzliche Argumente für die Completion.
    Returns:
        str: Antwort des LLM.
    """
    if not self.client.use_async:
        self.client.use_async = True
    return await self.client.achat_completion(messages, **kwargs)

chat(messages, **kwargs)

Führt eine Chat-Completion mit dem LLM aus.

Parameters:

Name Type Description Default
messages List[Dict[str, str]]

Nachrichtenverlauf für die LLM-Kommunikation.

 required
**kwargs Any

Zusätzliche Argumente for die Completion.

 {}

Returns: str: Antwort des LLM.

Source code in modul_anerkennung/llm_interface.py 
27
28
29
30
31
32
33
34
35
36
37
def chat(self, messages: List[Dict[str, str]], **kwargs: Any) -> str:
    """
    Führt eine Chat-Completion mit dem LLM aus.

    Args:
        messages (List[Dict[str, str]]): Nachrichtenverlauf für die LLM-Kommunikation.
        **kwargs: Zusätzliche Argumente for die Completion.
    Returns:
        str: Antwort des LLM.
    """
    return self.client.chat_completion(messages, **kwargs)

extract_json(text, model_class)

Extrahiert JSON aus einem Text und validiert es gegen eine Pydantic-Klasse.

Parameters:

Name Type Description Default
text str

Der Text, der JSON enthält.

 required
model_class Type[T]

Die Pydantic-Klasse zur Validierung.

 required

Returns:

Name Type Description
T T

Die validierte Instanz der model_class.
Source code in modul_anerkennung/llm_interface.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def extract_json(self, text: str, model_class: Type[T]) -> T:
    """
    Extrahiert JSON aus einem Text und validiert es gegen eine Pydantic-Klasse.

    Args:
        text (str): Der Text, der JSON enthält.
        model_class (Type[T]): Die Pydantic-Klasse zur Validierung.

    Returns:
        T: Die validierte Instanz der model_class.
    """
    # Suche nach dem ersten { und dem letzten }
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if not match:
        raise ValueError(f"Kein JSON im LLM-Output gefunden: {text}")

    json_str = match.group(0)
    try:
        data = json.loads(json_str)
        return model_class.model_validate(data)
    except (json.JSONDecodeError, Exception) as e:
        raise ValueError(
            f"Fehler beim Parsen oder Validieren des JSON: {e}\nRaw JSON: {json_str}"
        )