Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in
Toggle navigation
Menu
Open sidebar
math_tutor_dev
public_math_tutor
Commits
bd1e3cb2
Commit
bd1e3cb2
authored
Apr 29, 2026
by
Kantz
Browse files
auswahl des LLM providers in .env verschoben
parent
97edc264
Changes
7
Hide whitespace changes
Inline
Side-by-side
math-tutor/backend/.env-example
View file @
bd1e3cb2
...
...
@@ -9,6 +9,7 @@ FRONTEND_URL="http://localhost:5173"
ORCHESTRATOR="tutor" # "tutor" or "qa"
RETRIEVAL_IMPL="child" # "child" or "subsection"
LLM_PROVIDER="openai" # "openai", "mistral", or "ollama"
OPENAI_BASE_URL="https://chat-ai.academiccloud.de/v1/"
OPENAI_API_KEY=""
...
...
math-tutor/backend/app/LLM_services/decision_LLM.py
View file @
bd1e3cb2
"""Deprecated legacy decision LLM module.
The active tutor orchestrator no longer calls this module. Keep it only for
manual legacy tests until it can be removed.
"""
import
warnings
from
app.deterministic_services
import
llm_client
warnings
.
warn
(
"app.LLM_services.decision_LLM is deprecated and no longer used by the tutor orchestrator."
,
DeprecationWarning
,
stacklevel
=
2
,
)
def
context_decision
(
needs_more_context
:
bool
,
reason
:
str
)
->
dict
:
# Kannst auch einfach nur return {"needs_more_context": needs_more_context, "reason": reason}
...
...
math-tutor/backend/app/LLM_services/math_intent_LLM.py
View file @
bd1e3cb2
"""Deprecated legacy math-intent LLM module.
The active tutor orchestrator no longer calls this module. Keep it only for
manual legacy tests until it can be removed.
"""
import
warnings
import
sympy
as
sp
from
app.deterministic_services
import
llm_client
warnings
.
warn
(
"app.LLM_services.math_intent_LLM is deprecated and no longer used by the tutor orchestrator."
,
DeprecationWarning
,
stacklevel
=
2
,
)
def
sympy_solve
(
task
:
str
,
input
:
str
,
symbols
:
list
[
str
]
|
None
=
None
)
->
str
:
"""
...
...
math-tutor/backend/app/config.py
View file @
bd1e3cb2
...
...
@@ -7,6 +7,8 @@ from typing import Optional
load_dotenv
()
SUPPORTED_LLM_PROVIDERS
=
{
"openai"
,
"mistral"
,
"ollama"
}
class
EmbeddingSettings
(
BaseModel
):
embedding_type
:
str
# "openai-like" oder "sentence-transformer"
...
...
@@ -27,6 +29,21 @@ def get_retrieval_impl() -> str:
return
"child"
def
get_llm_provider
()
->
str
:
value
=
os
.
getenv
(
"LLM_PROVIDER"
)
if
value
is
None
or
not
value
.
strip
():
supported
=
", "
.
join
(
sorted
(
SUPPORTED_LLM_PROVIDERS
))
raise
ValueError
(
f
"Missing LLM_PROVIDER. Expected one of:
{
supported
}
"
)
provider
=
value
.
strip
().
lower
()
if
provider
not
in
SUPPORTED_LLM_PROVIDERS
:
supported
=
", "
.
join
(
sorted
(
SUPPORTED_LLM_PROVIDERS
))
raise
ValueError
(
f
"Unsupported LLM_PROVIDER:
{
value
}
. Expected one of:
{
supported
}
"
)
return
provider
def
get_embedding_settings
()
->
EmbeddingSettings
:
embedding_type
=
os
.
getenv
(
"EMBEDDING_TYPE"
,
"openai-like"
)
if
embedding_type
==
"sentence-transformer"
:
...
...
math-tutor/backend/app/deterministic_services/llm_client.py
View file @
bd1e3cb2
import
inspect
import
json
import
warnings
from
datetime
import
date
from
typing
import
Any
,
Callable
...
...
@@ -58,11 +59,74 @@ def _record_call(result: dict, tokens: int | None = None) -> dict:
llm_quota
.
record_usage
(
pg_url
,
date
.
today
(),
calls
=
1
,
tokens
=
token_count
)
return
result
def
_chat_openai
(
messages
:
list
[
dict
])
->
dict
:
def
_warn_deprecated_provider_flags
(
use_ollama
:
bool
,
use_mistral
:
bool
)
->
None
:
if
use_ollama
or
use_mistral
:
warnings
.
warn
(
"use_ollama and use_mistral are deprecated and ignored. "
"Set LLM_PROVIDER in the environment instead."
,
DeprecationWarning
,
stacklevel
=
3
,
)
def
_warn_deprecated_tools
()
->
None
:
warnings
.
warn
(
"LLM toolcalling via llm_client.chat(..., tools=...) is deprecated."
,
DeprecationWarning
,
stacklevel
=
3
,
)
def
_ensure_within_llm_quota
()
->
None
:
quota_settings
=
config
.
get_llm_quota_settings
()
llm_quota
.
ensure_within_limits
(
config
.
get_postgres_url
(),
date
.
today
(),
call_limit
=
quota_settings
.
daily_call_limit
,
token_limit
=
quota_settings
.
daily_token_limit
,
add_calls
=
1
,
add_tokens
=
0
,
)
def
_quota_tracked_chat
(
chat_func
:
Callable
[[],
dict
])
->
dict
:
_ensure_within_llm_quota
()
try
:
result
=
chat_func
()
except
Exception
:
_record_call
({
"raw"
:
None
},
tokens
=
0
)
raise
if
result
:
return
_record_call
(
result
)
_record_call
({
"raw"
:
None
},
tokens
=
0
)
return
result
def
_require_openai_chat_settings
()
->
config
.
OpenAIChatSettings
:
settings
=
config
.
get_openai_chat_settings
()
if
not
settings
:
return
{}
raise
ValueError
(
"LLM_PROVIDER=openai requires OPENAI_CHAT_MODEL, OPENAI_BASE_URL, "
"and OPENAI_API_KEY"
)
return
settings
def
_require_mistral_chat_settings
()
->
config
.
MistralChatSettings
:
settings
=
config
.
get_mistral_chat_settings
()
if
not
settings
:
raise
ValueError
(
"LLM_PROVIDER=mistral requires MISTRAL_CHAT_MODEL and MISTRAL_API_KEY"
)
return
settings
def
_chat_openai
(
messages
:
list
[
dict
],
settings
:
config
.
OpenAIChatSettings
|
None
=
None
,
)
->
dict
:
settings
=
settings
or
_require_openai_chat_settings
()
timeout
=
settings
.
timeout
or
60.0
client
=
OpenAI
(
api_key
=
settings
.
api_key
,
base_url
=
settings
.
base_url
,
timeout
=
timeout
)
...
...
@@ -74,11 +138,11 @@ def _chat_openai(messages: list[dict]) -> dict:
return
{
"raw"
:
response
,
"message"
:
message
}
def
_chat_mistral
(
messages
:
list
[
dict
])
->
dict
:
settings
=
config
.
get_mistral_chat_settings
()
if
not
settings
:
return
{}
def
_chat_mistral
(
messages
:
list
[
dict
],
settings
:
config
.
MistralChatSettings
|
None
=
None
,
)
->
dict
:
settings
=
settings
or
_require_mistral_chat_settings
()
kwargs
:
dict
[
str
,
Any
]
=
{
"model"
:
settings
.
model
,
"messages"
:
messages
,
...
...
@@ -97,89 +161,10 @@ def _chat_mistral(messages: list[dict]) -> dict:
return
{
"raw"
:
response
,
"message"
:
message
}
def
get_message_content
(
result
:
dict
|
object
)
->
str
:
message
=
result
.
get
(
"message"
)
if
isinstance
(
result
,
dict
)
else
result
if
isinstance
(
message
,
dict
):
return
message
.
get
(
"content"
,
""
)
or
""
if
hasattr
(
message
,
"content"
):
return
getattr
(
message
,
"content"
)
or
""
return
""
def
chat
(
def
_chat_ollama
(
messages
:
list
[
dict
],
tools
:
list
[
Callable
[...,
Any
]]
|
None
=
None
,
use_ollama
:
bool
=
False
,
use_mistral
:
bool
=
False
,
)
->
dict
:
quota_settings
=
config
.
get_llm_quota_settings
()
pg_url
=
config
.
get_postgres_url
()
if
use_mistral
and
tools
:
raise
RuntimeError
(
"Mistral chat is currently only implemented for calls without tools."
)
if
use_mistral
:
mistral_settings
=
config
.
get_mistral_chat_settings
()
if
not
mistral_settings
:
return
{}
llm_quota
.
ensure_within_limits
(
pg_url
,
date
.
today
(),
call_limit
=
quota_settings
.
daily_call_limit
,
token_limit
=
quota_settings
.
daily_token_limit
,
add_calls
=
1
,
add_tokens
=
0
,
)
try
:
mistral_result
=
_chat_mistral
(
messages
)
except
Exception
:
_record_call
({
"raw"
:
None
},
tokens
=
0
)
raise
if
mistral_result
:
return
_record_call
(
mistral_result
)
_record_call
({
"raw"
:
None
},
tokens
=
0
)
return
mistral_result
if
not
tools
and
not
use_ollama
:
openai_settings
=
config
.
get_openai_chat_settings
()
if
openai_settings
:
llm_quota
.
ensure_within_limits
(
pg_url
,
date
.
today
(),
call_limit
=
quota_settings
.
daily_call_limit
,
token_limit
=
quota_settings
.
daily_token_limit
,
add_calls
=
1
,
add_tokens
=
0
,
)
try
:
openai_result
=
_chat_openai
(
messages
)
except
Exception
:
_record_call
({
"raw"
:
None
},
tokens
=
0
)
raise
if
openai_result
:
return
_record_call
(
openai_result
)
_record_call
({
"raw"
:
None
},
tokens
=
0
)
return
openai_result
mistral_settings
=
config
.
get_mistral_chat_settings
()
if
mistral_settings
:
llm_quota
.
ensure_within_limits
(
pg_url
,
date
.
today
(),
call_limit
=
quota_settings
.
daily_call_limit
,
token_limit
=
quota_settings
.
daily_token_limit
,
add_calls
=
1
,
add_tokens
=
0
,
)
try
:
mistral_result
=
_chat_mistral
(
messages
)
except
Exception
:
_record_call
({
"raw"
:
None
},
tokens
=
0
)
raise
if
mistral_result
:
return
_record_call
(
mistral_result
)
_record_call
({
"raw"
:
None
},
tokens
=
0
)
return
mistral_result
settings
=
config
.
get_ollama_settings
()
client
=
ollama
.
Client
(
host
=
settings
.
base_url
,
timeout
=
settings
.
timeout
)
...
...
@@ -202,6 +187,38 @@ def chat(
return
{
"raw"
:
response
,
"message"
:
_extract_message
(
response
)}
def
get_message_content
(
result
:
dict
|
object
)
->
str
:
message
=
result
.
get
(
"message"
)
if
isinstance
(
result
,
dict
)
else
result
if
isinstance
(
message
,
dict
):
return
message
.
get
(
"content"
,
""
)
or
""
if
hasattr
(
message
,
"content"
):
return
getattr
(
message
,
"content"
)
or
""
return
""
def
chat
(
messages
:
list
[
dict
],
tools
:
list
[
Callable
[...,
Any
]]
|
None
=
None
,
use_ollama
:
bool
=
False
,
use_mistral
:
bool
=
False
,
)
->
dict
:
if
tools
:
_warn_deprecated_tools
()
_warn_deprecated_provider_flags
(
use_ollama
,
use_mistral
)
provider
=
config
.
get_llm_provider
()
if
provider
==
"openai"
:
settings
=
_require_openai_chat_settings
()
return
_quota_tracked_chat
(
lambda
:
_chat_openai
(
messages
,
settings
))
if
provider
==
"mistral"
:
settings
=
_require_mistral_chat_settings
()
return
_quota_tracked_chat
(
lambda
:
_chat_mistral
(
messages
,
settings
))
if
provider
==
"ollama"
:
return
_chat_ollama
(
messages
,
tools
=
tools
)
raise
ValueError
(
f
"Unsupported LLM_PROVIDER:
{
provider
}
"
)
def
chat_with_tools
(
messages
:
list
[
dict
],
tools
:
list
[
Callable
[...,
Any
]],
...
...
@@ -209,6 +226,12 @@ def chat_with_tools(
use_mistral
:
bool
=
False
,
return_after_tools
:
bool
=
False
,
)
->
tuple
[
dict
,
list
[
dict
[
str
,
Any
]]]:
"""Deprecated legacy wrapper around model tool calls."""
warnings
.
warn
(
"llm_client.chat_with_tools(...) is deprecated."
,
DeprecationWarning
,
stacklevel
=
2
,
)
tool_map
=
{
tool
.
__name__
:
tool
for
tool
in
tools
}
result
=
chat
(
messages
=
messages
,
...
...
@@ -235,6 +258,7 @@ def _apply_tool_calls(
messages
:
list
[
dict
],
tool_map
:
dict
[
str
,
Callable
[...,
Any
]],
)
->
list
[
dict
[
str
,
Any
]]:
"""Deprecated legacy helper for chat_with_tools."""
message
=
result
.
get
(
"message"
)
if
isinstance
(
result
,
dict
)
else
result
tool_calls
=
_extract_tool_calls
(
message
)
outputs
:
list
[
dict
[
str
,
Any
]]
=
[]
...
...
@@ -262,6 +286,7 @@ def _apply_tool_calls(
def
_extract_tool_calls
(
message
:
object
)
->
list
:
"""Deprecated legacy helper for chat_with_tools."""
if
isinstance
(
message
,
dict
):
return
message
.
get
(
"tool_calls"
)
or
[]
if
hasattr
(
message
,
"tool_calls"
):
...
...
@@ -270,6 +295,7 @@ def _extract_tool_calls(message: object) -> list:
def
_tool_call_name_args
(
call
:
object
)
->
tuple
[
str
,
dict
[
str
,
Any
]]:
"""Deprecated legacy helper for chat_with_tools."""
if
isinstance
(
call
,
dict
):
function
=
call
.
get
(
"function"
)
or
{}
name
=
function
.
get
(
"name"
)
or
""
...
...
@@ -282,6 +308,7 @@ def _tool_call_name_args(call: object) -> tuple[str, dict[str, Any]]:
def
_parse_tool_arguments
(
arguments
:
object
)
->
dict
[
str
,
Any
]:
"""Deprecated legacy helper for chat_with_tools."""
if
isinstance
(
arguments
,
dict
):
return
arguments
if
isinstance
(
arguments
,
str
)
and
arguments
.
strip
():
...
...
math-tutor/backend/app/deterministic_services/orchestrators/orchestrator_tutor.py
View file @
bd1e3cb2
from
__future__
import
annotations
from
app.LLM_services
import
decision_LLM
,
open_hint_LLM
,
math_inte
nt_LLM
,
solver_LLM
from
app.LLM_services
import
open_hi
nt_LLM
,
solver_LLM
from
app.deterministic_services
import
context_store
from
app.deterministic_services.orchestrators
import
orchestrator_base
as
base
...
...
@@ -8,21 +8,12 @@ from app.deterministic_services.orchestrators import orchestrator_base as base
def
_on_bootstrap
(
state
:
base
.
ChatState
,
query_text
:
str
)
->
None
:
base
.
bootstrap_retrieval
(
state
.
sheet
,
query_text
,
state
.
tool_log
)
math_solution
=
base
.
log_timed_call
(
state
.
tool_log
,
"math_intent_LLM"
,
{
"query"
:
query_text
},
lambda
:
math_intent_LLM
.
solve_with_tools
(
query_text
),
)
if
math_solution
:
context_store
.
add_math_solution
(
state
.
sheet
,
math_solution
)
# Sonstiges bei jedem Aufruf
def
_on_turn_logic
(
state
:
base
.
ChatState
)
->
None
:
sheet_text
=
context_store
.
format_sheet
(
state
.
sheet
)
if
state
.
new_chat
:
sheet_text
=
context_store
.
format_sheet
(
state
.
sheet
)
llm_solution
=
base
.
log_timed_call
(
state
.
tool_log
,
"LLM_Solution"
,
...
...
@@ -30,19 +21,6 @@ def _on_turn_logic(state: base.ChatState) -> None:
lambda
:
solver_LLM
.
solve_question
(
state
.
last_user
,
sheet_text
),
)
context_store
.
add_LLM_solution
(
state
.
sheet
,
llm_solution
)
return
decision
=
base
.
log_timed_call
(
state
.
tool_log
,
"decision"
,
{
"sheet"
:
sheet_text
},
lambda
:
decision_LLM
.
needs_more_context
(
sheet_text
),
)
context_store
.
add_decision
(
state
.
sheet
,
decision
)
if
decision
.
get
(
"needs_more_context"
):
full_query
=
"
\n
"
.
join
(
base
.
extract_user_messages
(
state
.
messages
))
_on_bootstrap
(
state
,
full_query
)
# Antwort generieren
...
...
math-tutor/backend/test/test_llm_provider.py
0 → 100644
View file @
bd1e3cb2
import
importlib.util
import
os
import
sys
import
types
import
unittest
import
warnings
from
unittest.mock
import
patch
os
.
environ
.
setdefault
(
"EMBEDDING_TYPE"
,
"sentence-transformer"
)
os
.
environ
.
setdefault
(
"EMBEDDING_DIM"
,
"512"
)
psycopg_stub
=
types
.
ModuleType
(
"psycopg"
)
psycopg_rows_stub
=
types
.
ModuleType
(
"psycopg.rows"
)
psycopg_rows_stub
.
dict_row
=
object
()
psycopg_stub
.
rows
=
psycopg_rows_stub
pgvector_stub
=
types
.
ModuleType
(
"pgvector"
)
pgvector_psycopg_stub
=
types
.
ModuleType
(
"pgvector.psycopg"
)
pgvector_stub
.
Vector
=
list
pgvector_psycopg_stub
.
register_vector
=
lambda
conn
:
None
ollama_stub
=
types
.
ModuleType
(
"ollama"
)
ollama_stub
.
Client
=
object
openai_stub
=
types
.
ModuleType
(
"openai"
)
openai_stub
.
OpenAI
=
object
mistralai_stub
=
types
.
ModuleType
(
"mistralai"
)
mistralai_client_stub
=
types
.
ModuleType
(
"mistralai.client"
)
mistralai_client_stub
.
Mistral
=
object
sentence_transformers_stub
=
types
.
ModuleType
(
"sentence_transformers"
)
class
_SentenceTransformerStub
:
def
__init__
(
self
,
*
args
,
**
kwargs
)
->
None
:
self
.
max_seq_length
=
None
def
encode
(
self
,
*
args
,
**
kwargs
)
->
list
[
list
[
float
]]:
return
[[
0.0
]]
sentence_transformers_stub
.
SentenceTransformer
=
_SentenceTransformerStub
if
importlib
.
util
.
find_spec
(
"psycopg"
)
is
None
:
sys
.
modules
.
setdefault
(
"psycopg"
,
psycopg_stub
)
sys
.
modules
.
setdefault
(
"psycopg.rows"
,
psycopg_rows_stub
)
if
importlib
.
util
.
find_spec
(
"pgvector"
)
is
None
:
sys
.
modules
.
setdefault
(
"pgvector"
,
pgvector_stub
)
sys
.
modules
.
setdefault
(
"pgvector.psycopg"
,
pgvector_psycopg_stub
)
if
importlib
.
util
.
find_spec
(
"ollama"
)
is
None
:
sys
.
modules
.
setdefault
(
"ollama"
,
ollama_stub
)
if
importlib
.
util
.
find_spec
(
"openai"
)
is
None
:
sys
.
modules
.
setdefault
(
"openai"
,
openai_stub
)
if
importlib
.
util
.
find_spec
(
"mistralai"
)
is
None
:
sys
.
modules
.
setdefault
(
"mistralai"
,
mistralai_stub
)
sys
.
modules
.
setdefault
(
"mistralai.client"
,
mistralai_client_stub
)
else
:
try
:
import
mistralai.client
as
installed_mistralai_client
except
ImportError
:
sys
.
modules
.
setdefault
(
"mistralai.client"
,
mistralai_client_stub
)
else
:
if
not
hasattr
(
installed_mistralai_client
,
"Mistral"
):
installed_mistralai_client
.
Mistral
=
object
if
importlib
.
util
.
find_spec
(
"sentence_transformers"
)
is
None
:
sys
.
modules
.
setdefault
(
"sentence_transformers"
,
sentence_transformers_stub
)
from
app
import
config
from
app.deterministic_services
import
llm_client
from
app.deterministic_services.orchestrators
import
orchestrator_tutor
from
app.deterministic_services.orchestrators.orchestrator_base
import
ChatState
MESSAGES
=
[{
"role"
:
"user"
,
"content"
:
"Hallo"
}]
def
_dummy_tool
()
->
str
:
return
"ok"
class
LLMProviderConfigTest
(
unittest
.
TestCase
):
def
test_get_llm_provider_accepts_supported_values
(
self
)
->
None
:
for
provider
in
(
"openai"
,
"mistral"
,
"ollama"
):
with
self
.
subTest
(
provider
=
provider
),
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
provider
},
clear
=
True
):
self
.
assertEqual
(
config
.
get_llm_provider
(),
provider
)
def
test_get_llm_provider_normalizes_case_and_space
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
" OpenAI "
},
clear
=
True
):
self
.
assertEqual
(
config
.
get_llm_provider
(),
"openai"
)
def
test_get_llm_provider_rejects_missing_value
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{},
clear
=
True
):
with
self
.
assertRaisesRegex
(
ValueError
,
"Missing LLM_PROVIDER"
):
config
.
get_llm_provider
()
def
test_get_llm_provider_rejects_unknown_value
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"anthropic"
},
clear
=
True
):
with
self
.
assertRaisesRegex
(
ValueError
,
"Unsupported LLM_PROVIDER"
):
config
.
get_llm_provider
()
class
LLMClientProviderTest
(
unittest
.
TestCase
):
def
test_chat_uses_only_openai_provider
(
self
)
->
None
:
settings
=
object
()
expected
=
{
"raw"
:
object
(),
"message"
:
{
"content"
:
"openai"
}}
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"openai"
}),
patch
.
object
(
llm_client
,
"_require_openai_chat_settings"
,
return_value
=
settings
),
patch
.
object
(
llm_client
,
"_ensure_within_llm_quota"
),
patch
.
object
(
llm_client
,
"_record_call"
,
side_effect
=
lambda
result
,
tokens
=
None
:
result
),
patch
.
object
(
llm_client
,
"_chat_openai"
,
return_value
=
expected
)
as
openai_chat
,
patch
.
object
(
llm_client
,
"_chat_mistral"
)
as
mistral_chat
,
patch
.
object
(
llm_client
,
"_chat_ollama"
)
as
ollama_chat
:
result
=
llm_client
.
chat
(
MESSAGES
)
self
.
assertEqual
(
result
,
expected
)
openai_chat
.
assert_called_once_with
(
MESSAGES
,
settings
)
mistral_chat
.
assert_not_called
()
ollama_chat
.
assert_not_called
()
def
test_chat_uses_only_mistral_provider
(
self
)
->
None
:
settings
=
object
()
expected
=
{
"raw"
:
object
(),
"message"
:
{
"content"
:
"mistral"
}}
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"mistral"
}),
patch
.
object
(
llm_client
,
"_require_mistral_chat_settings"
,
return_value
=
settings
),
patch
.
object
(
llm_client
,
"_ensure_within_llm_quota"
),
patch
.
object
(
llm_client
,
"_record_call"
,
side_effect
=
lambda
result
,
tokens
=
None
:
result
),
patch
.
object
(
llm_client
,
"_chat_openai"
)
as
openai_chat
,
patch
.
object
(
llm_client
,
"_chat_mistral"
,
return_value
=
expected
)
as
mistral_chat
,
patch
.
object
(
llm_client
,
"_chat_ollama"
)
as
ollama_chat
:
result
=
llm_client
.
chat
(
MESSAGES
)
self
.
assertEqual
(
result
,
expected
)
openai_chat
.
assert_not_called
()
mistral_chat
.
assert_called_once_with
(
MESSAGES
,
settings
)
ollama_chat
.
assert_not_called
()
def
test_chat_uses_only_ollama_provider
(
self
)
->
None
:
expected
=
{
"raw"
:
object
(),
"message"
:
{
"content"
:
"ollama"
}}
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"ollama"
}),
patch
.
object
(
llm_client
,
"_chat_openai"
)
as
openai_chat
,
patch
.
object
(
llm_client
,
"_chat_mistral"
)
as
mistral_chat
,
patch
.
object
(
llm_client
,
"_chat_ollama"
,
return_value
=
expected
)
as
ollama_chat
:
result
=
llm_client
.
chat
(
MESSAGES
)
self
.
assertEqual
(
result
,
expected
)
openai_chat
.
assert_not_called
()
mistral_chat
.
assert_not_called
()
ollama_chat
.
assert_called_once_with
(
MESSAGES
,
tools
=
None
)
def
test_chat_deprecated_provider_flags_are_ignored
(
self
)
->
None
:
settings
=
object
()
expected
=
{
"raw"
:
object
(),
"message"
:
{
"content"
:
"openai"
}}
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"openai"
}),
patch
.
object
(
llm_client
,
"_require_openai_chat_settings"
,
return_value
=
settings
),
patch
.
object
(
llm_client
,
"_ensure_within_llm_quota"
),
patch
.
object
(
llm_client
,
"_record_call"
,
side_effect
=
lambda
result
,
tokens
=
None
:
result
),
patch
.
object
(
llm_client
,
"_chat_openai"
,
return_value
=
expected
)
as
openai_chat
,
patch
.
object
(
llm_client
,
"_chat_ollama"
)
as
ollama_chat
:
with
self
.
assertWarns
(
DeprecationWarning
):
result
=
llm_client
.
chat
(
MESSAGES
,
use_ollama
=
True
,
use_mistral
=
True
)
self
.
assertEqual
(
result
,
expected
)
openai_chat
.
assert_called_once_with
(
MESSAGES
,
settings
)
ollama_chat
.
assert_not_called
()
def
test_selected_provider_config_error_happens_before_quota
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"openai"
},
clear
=
True
),
patch
.
object
(
llm_client
,
"_ensure_within_llm_quota"
)
as
ensure_quota
:
with
self
.
assertRaisesRegex
(
ValueError
,
"LLM_PROVIDER=openai requires"
):
llm_client
.
chat
(
MESSAGES
)
ensure_quota
.
assert_not_called
()
def
test_chat_without_provider_does_not_fallback
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{},
clear
=
True
),
patch
.
object
(
llm_client
,
"_chat_openai"
)
as
openai_chat
,
patch
.
object
(
llm_client
,
"_chat_mistral"
)
as
mistral_chat
,
patch
.
object
(
llm_client
,
"_chat_ollama"
)
as
ollama_chat
:
with
self
.
assertRaisesRegex
(
ValueError
,
"Missing LLM_PROVIDER"
):
llm_client
.
chat
(
MESSAGES
)
openai_chat
.
assert_not_called
()
mistral_chat
.
assert_not_called
()
ollama_chat
.
assert_not_called
()
def
test_chat_tools_argument_is_deprecated
(
self
)
->
None
:
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"openai"
}):
with
warnings
.
catch_warnings
(
record
=
True
)
as
caught
:
warnings
.
simplefilter
(
"always"
)
with
self
.
assertRaisesRegex
(
RuntimeError
,
"toolcalling"
):
llm_client
.
chat
(
MESSAGES
,
tools
=
[
_dummy_tool
])
messages
=
[
str
(
warning
.
message
)
for
warning
in
caught
]
self
.
assertTrue
(
any
(
"tools"
in
message
for
message
in
messages
))
def
test_chat_with_tools_is_deprecated
(
self
)
->
None
:
response
=
{
"raw"
:
object
(),
"message"
:
{
"content"
:
"ok"
}}
with
patch
.
dict
(
os
.
environ
,
{
"LLM_PROVIDER"
:
"ollama"
}),
patch
.
object
(
llm_client
,
"_chat_ollama"
,
return_value
=
response
):
with
warnings
.
catch_warnings
(
record
=
True
)
as
caught
:
warnings
.
simplefilter
(
"always"
)
result
,
tool_outputs
=
llm_client
.
chat_with_tools
(
MESSAGES
[:],
[
_dummy_tool
],
return_after_tools
=
True
)
messages
=
[
str
(
warning
.
message
)
for
warning
in
caught
]
self
.
assertEqual
(
result
,
response
)
self
.
assertEqual
(
tool_outputs
,
[])
self
.
assertTrue
(
any
(
"chat_with_tools"
in
message
for
message
in
messages
))
self
.
assertTrue
(
any
(
"tools"
in
message
for
message
in
messages
))
class
TutorOrchestratorLegacyModuleTest
(
unittest
.
TestCase
):
def
test_tutor_orchestrator_does_not_import_legacy_llm_modules
(
self
)
->
None
:
self
.
assertFalse
(
hasattr
(
orchestrator_tutor
,
"decision_LLM"
))
self
.
assertFalse
(
hasattr
(
orchestrator_tutor
,
"math_intent_LLM"
))
def
test_bootstrap_only_runs_retrieval
(
self
)
->
None
:
state
=
ChatState
(
messages
=
MESSAGES
[:],
draft
=
None
,
chat_id
=
"chat-1"
,
new_chat
=
True
,
sheet
=
{},
tool_log
=
[],
last_user
=
"Hallo"
,
)
with
patch
.
object
(
orchestrator_tutor
.
base
,
"bootstrap_retrieval"
)
as
bootstrap
:
orchestrator_tutor
.
_on_bootstrap
(
state
,
"Hallo"
)
bootstrap
.
assert_called_once_with
(
state
.
sheet
,
"Hallo"
,
state
.
tool_log
)
self
.
assertEqual
(
state
.
tool_log
,
[])
def
test_non_new_turn_does_not_run_decision_llm
(
self
)
->
None
:
state
=
ChatState
(
messages
=
MESSAGES
[:]
+
[{
"role"
:
"assistant"
,
"content"
:
"Antwort"
}],
draft
=
None
,
chat_id
=
"chat-1"
,
new_chat
=
False
,
sheet
=
{
"history"
:
[]},
tool_log
=
[],
last_user
=
"Hallo"
,
)
with
patch
.
object
(
orchestrator_tutor
.
solver_LLM
,
"solve_question"
)
as
solver
:
orchestrator_tutor
.
_on_turn_logic
(
state
)
solver
.
assert_not_called
()
self
.
assertEqual
(
state
.
tool_log
,
[])
if
__name__
==
"__main__"
:
unittest
.
main
()
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment