Langfuse: Add comprehensive error handling and tests #551
Conversation
- Added try-catch blocks to prevent Langfuse failures from blocking main flow - Ensure tracer is always initialized (disabled if no credentials) - Added graceful degradation for all tracer methods - Created comprehensive test suite covering all failure scenarios - Tests verify OpenAI responses work regardless of Langfuse state Fixes issue where Langfuse connection failures could block API responses.
vprashrex
requested review from
AkhileshNegi,
Prajna1999,
kartpop and
nishika26
📝 WalkthroughWalkthroughAdds defensive failure handling to LangfuseTracer: introduces a Changes
Sequence Diagram(s)(omitted — changes are defensive wrappers and tests; no new multi-component control flow requiring visualization) Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
| "[LangfuseTracer] Langfuse tracing disabled due to missing credentials" | ||
| ) | ||
| logger.info( | ||
| f"[LangfuseTracer] Langfuse tracing enabled | session_id={self.session_id}" |
There was a problem hiding this comment.
this is somehow still populating in the logs even though the credentials are revoked
| # ============================================================================= | ||
| # Fixtures | ||
| # ============================================================================= |
There was a problem hiding this comment.
lots of unnecessary formatting
Prajna1999
left a comment
Prajna1999
left a comment
There was a problem hiding this comment.
The overall sudo_code is
if langfuse_fail: continue
I think there are far simpler ways to implement this logic instead of having individual try_catch blocks for oneliner method invocations.
| - No credentials (None) | ||
| - Incomplete credentials (missing keys) | ||
| - Langfuse initialization exception | ||
| - fetch_traces failure (non-critical) |
There was a problem hiding this comment.
nitpick: isn't fetch_traces failure is critical? Like am I missing some context here
| if not self.langfuse.enabled: | ||
| return | ||
| try: | ||
| self.langfuse.flush() |
There was a problem hiding this comment.
Is try_catch really needed for one line method invocation? May be keep the one liner langfuse.flush() wherever it needs to be invoked without this extra abstraction.
|
|
||
| def safe_langfuse_op(op: Callable, *args, **kwargs): | ||
| try: | ||
| return op(*args, **kwargs) |
There was a problem hiding this comment.
ditto as above. Does really a oneliner require its own try_catch
| if self.trace: | ||
| self.trace.update(tags=tags, output=output) | ||
| if not self.langfuse.enabled or not self.trace: | ||
| return |
Address PR review feedback - replace multiple try/catch blocks with single _langfuse_call wrapper that auto-disables on first failure.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
backend/app/core/langfuse/langfuse.py (3)
1-5: CI formatting failure — run Black on this file.Black reformatting failed in CI for this file; please re-run Black to match the configured style.
141-168: Add type hints to decorator, wrapper, andlangfuse_callfunctions.The
observe_llm_executiondecorator is missing critical type annotations: the function lacks a return type, thedecoratorparameter type forfuncis too generic, thewrapperfunction has no return type and**kwargslacks type annotation, and the nestedlangfuse_callfunction has no parameter or return type hints.♻️ Proposed fix
-def observe_llm_execution( - session_id: str | None = None, - credentials: dict | None = None, -): +def observe_llm_execution( + session_id: str | None = None, + credentials: dict | None = None, +) -> Callable[ + [Callable[..., tuple[LLMCallResponse | None, str | None]]], + Callable[..., tuple[LLMCallResponse | None, str | None]], +]: @@ - def decorator(func: Callable) -> Callable: + def decorator( + func: Callable[..., tuple[LLMCallResponse | None, str | None]] + ) -> Callable[..., tuple[LLMCallResponse | None, str | None]]: @@ - def wrapper( - completion_config: NativeCompletionConfig, query: QueryParams, **kwargs - ): + def wrapper( + completion_config: NativeCompletionConfig, + query: QueryParams, + **kwargs: Any, + ) -> tuple[LLMCallResponse | None, str | None]: @@ - def langfuse_call(fn, *args, **kwargs): + def langfuse_call( + fn: Callable[..., Any], *args: Any, **kwargs: Any + ) -> Any | None:Also applies to: 183-189
72-138: Add explicit-> Nonereturn types to all tracer methods.These methods lack return type annotations required by coding guidelines. All parameter types are present; only return types are missing.
♻️ Proposed fix
- def start_trace( + def start_trace( self, name: str, input: Dict[str, Any], metadata: Optional[Dict[str, Any]] = None, tags: list[str] | None = None, - ): + ) -> None: @@ - def start_generation( + def start_generation( self, name: str, input: Dict[str, Any], metadata: Optional[Dict[str, Any]] = None, - ): + ) -> None: @@ - def end_generation( + def end_generation( self, output: Dict[str, Any], usage: Optional[Dict[str, Any]] = None, model: Optional[str] = None, - ): + ) -> None: @@ - def update_trace(self, tags: list[str], output: Dict[str, Any]): + def update_trace(self, tags: list[str], output: Dict[str, Any]) -> None: @@ - def log_error(self, error_message: str, response_id: Optional[str] = None): + def log_error(self, error_message: str, response_id: Optional[str] = None) -> None: @@ - def flush(self): + def flush(self) -> None:
🤖 Fix all issues with AI agents
In `@backend/app/core/langfuse/langfuse.py`:
- Around line 174-176: Update the logger.info calls that print Langfuse tracing
state to include the current function name as a prefix (e.g. "[<function_name>]
Tracing enabled ..."); specifically modify the logger.info line that logs
session_id and the subsequent logger calls around lines 191-192 to start with
that prefix. Locate the logger.info calls (the ones referencing session_id and
the later Langfuse logs) inside the function that contains them (the function
where these logger calls live—e.g., the enable_langfuse-related function) and
prepend the function name (either hard-coded with the function identifier or
generated via the function's __name__) to the log message; also update the
earlier similar log (around the earlier logger call near line 165) for
consistency. Ensure formatting remains f-string-compatible (e.g.
f"[{function_name}] Tracing enabled | session_id={session_id or 'auto'}").
- Around line 62-70: Add type hints and include the function-name prefix in the
log: annotate _langfuse_call with proper typing (e.g., add "from typing import
Callable, TypeVar, Optional, Any", define T = TypeVar('T'), and change the
signature to def _langfuse_call(self, fn: Callable[..., T], *args: Any,
**kwargs: Any) -> Optional[T]:), keep the existing self._failed handling and
return None where appropriate, and update the logger.warning message to include
the method name prefix (for example include
"LangfuseTracer._langfuse_call:{getattr(fn, '__name__', 'operation')}" at the
start of the log message) so logs follow the coding guideline.
In `@backend/app/tests/core/test_langfuse/test_langfuse_tracer.py`:
- Around line 1-5: This file fails CI formatting; run Black to reformat
backend/app/tests/core/test_langfuse/test_langfuse_tracer.py (or run black .)
and commit the updated file so imports, docstring, and overall formatting match
the project's Black configuration; no code changes required—just apply Black
formatting.
- Around line 14-36: The fixtures valid_credentials and assistant_mock currently
return concrete objects; change them to factory-style fixtures that return
callables which produce the data when invoked (i.e., valid_credentials -> a
function that returns a dict and assistant_mock -> a function that returns an
Assistant instance), update tests to call the factories by invoking
valid_credentials()() and assistant_mock()() (or simply calling the returned
callable) wherever they currently use the fixture values; ensure the fixture
names remain the same and reference the Assistant constructor signature used in
the snippet (Assistant with id, assistant_id, name, model, temperature,
instructions, vector_store_ids, max_num_results, project_id, organization_id).
| def _langfuse_call(self, fn, *args, **kwargs): | ||
| if self._failed: | ||
| return None | ||
| try: | ||
| return fn(*args, **kwargs) | ||
| except Exception as e: | ||
| logger.warning(f"[LangfuseTracer] {getattr(fn, '__name__', 'operation')} failed: {e}") | ||
| self._failed = True | ||
| return None |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n backend/app/core/langfuse/langfuse.py | head -100Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 3988
Add type hints and function-name log prefix in _langfuse_call.
Missing parameter and return type annotations, and the log message prefix must include the method name per coding guidelines.
♻️ Proposed fix
- def _langfuse_call(self, fn, *args, **kwargs):
+ def _langfuse_call(
+ self,
+ fn: Callable[..., Any],
+ *args: Any,
+ **kwargs: Any,
+ ) -> Any | None:
if self._failed:
return None
try:
return fn(*args, **kwargs)
except Exception as e:
- logger.warning(f"[LangfuseTracer] {getattr(fn, '__name__', 'operation')} failed: {e}")
+ logger.warning(
+ f"[LangfuseTracer._langfuse_call] {getattr(fn, '__name__', 'operation')} failed: {e}"
+ )
self._failed = True
return None🤖 Prompt for AI Agents
In `@backend/app/core/langfuse/langfuse.py` around lines 62 - 70, Add type hints
and include the function-name prefix in the log: annotate _langfuse_call with
proper typing (e.g., add "from typing import Callable, TypeVar, Optional, Any",
define T = TypeVar('T'), and change the signature to def _langfuse_call(self,
fn: Callable[..., T], *args: Any, **kwargs: Any) -> Optional[T]:), keep the
existing self._failed handling and return None where appropriate, and update the
logger.warning message to include the method name prefix (for example include
"LangfuseTracer._langfuse_call:{getattr(fn, '__name__', 'operation')}" at the
start of the log message) so logs follow the coding guideline.
| @pytest.fixture | ||
| def valid_credentials() -> dict: | ||
| return { | ||
| "public_key": "pk-test-123", | ||
| "secret_key": "sk-test-456", | ||
| "host": "https://langfuse.example.com", | ||
| } | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def assistant_mock() -> Assistant: | ||
| return Assistant( | ||
| id=123, | ||
| assistant_id="asst_test123", | ||
| name="Test Assistant", | ||
| model="gpt-4", | ||
| temperature=0.7, | ||
| instructions="You are a helpful assistant.", | ||
| vector_store_ids=["vs1"], | ||
| max_num_results=5, | ||
| project_id=1, | ||
| organization_id=1, | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
🧩 Analysis chain
🏁 Script executed:
# First, let's check if the file exists and examine its current content
cd backend/app/tests/core/test_langfuse || find . -name "test_langfuse_tracer.py" -type fRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 57
🏁 Script executed:
# Search for the test file in the repository
find . -name "test_langfuse_tracer.py" -type f 2>/dev/nullRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 137
🏁 Script executed:
# Let's explore the directory structure to find test files
find . -path "*/backend/app/tests/*" -name "*.py" -type f | head -20Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 1245
🏁 Script executed:
# Look for patterns of factory fixtures in existing test files
rg "def.*_factory\(\)" --type py -A 3 | head -50Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 57
🏁 Script executed:
# Check how the current fixtures are being used
rg "valid_credentials|assistant_mock" --type py | head -30Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 3589
🏁 Script executed:
# Let's read the actual test file content
cat -n ./backend/app/tests/core/test_langfuse/test_langfuse_tracer.pyRepository: ProjectTech4DevAI/kaapi-backend
Length of output: 26201
🏁 Script executed:
# Check for existing factory pattern examples in other test files
rg "@pytest.fixture" -A 8 backend/app/tests/ --type py | grep -A 8 "_factory" | head -50Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 57
🏁 Script executed:
# Look at conftest.py files to see how fixtures are structured
find ./backend/app/tests -name "conftest.py" -exec cat -n {} + | head -100Repository: ProjectTech4DevAI/kaapi-backend
Length of output: 3829
Convert fixtures to factory pattern as required by coding guidelines.
Test fixtures in backend/app/tests/ must follow the factory pattern. Currently, valid_credentials() and assistant_mock() return concrete objects directly. They should return callables that can generate test data, allowing flexibility for test variations.
♻️ Update fixtures to factory pattern
+from typing import Callable
+
`@pytest.fixture`
-def valid_credentials() -> dict:
- return {
- "public_key": "pk-test-123",
- "secret_key": "sk-test-456",
- "host": "https://langfuse.example.com",
- }
+def valid_credentials() -> Callable[..., dict]:
+ def _factory(
+ public_key: str = "pk-test-123",
+ secret_key: str = "sk-test-456",
+ host: str = "https://langfuse.example.com",
+ ) -> dict:
+ return {
+ "public_key": public_key,
+ "secret_key": secret_key,
+ "host": host,
+ }
+
+ return _factory
`@pytest.fixture`
-def assistant_mock() -> Assistant:
- return Assistant(
+def assistant_mock() -> Callable[..., Assistant]:
+ def _factory() -> Assistant:
+ return Assistant(
id=123,
assistant_id="asst_test123",
name="Test Assistant",
model="gpt-4",
temperature=0.7,
instructions="You are a helpful assistant.",
vector_store_ids=["vs1"],
max_num_results=5,
project_id=1,
organization_id=1,
- )
+ )
+
+ return _factoryUpdate all test methods to call these fixtures: valid_credentials() → valid_credentials()()
🤖 Prompt for AI Agents
In `@backend/app/tests/core/test_langfuse/test_langfuse_tracer.py` around lines 14
- 36, The fixtures valid_credentials and assistant_mock currently return
concrete objects; change them to factory-style fixtures that return callables
which produce the data when invoked (i.e., valid_credentials -> a function that
returns a dict and assistant_mock -> a function that returns an Assistant
instance), update tests to call the factories by invoking valid_credentials()()
and assistant_mock()() (or simply calling the returned callable) wherever they
currently use the fixture values; ensure the fixture names remain the same and
reference the Assistant constructor signature used in the snippet (Assistant
with id, assistant_id, name, model, temperature, instructions, vector_store_ids,
max_num_results, project_id, organization_id).
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/app/core/langfuse/langfuse.py (1)
162-166: Add type hints forwrapperand locallangfuse_call.Both functions lack required type annotations per coding guidelines. The
wrapperfunction's**kwargsparameter and return type are untyped; it returns a tuple of(LLMCallResponse | None, str | None). Thelangfuse_callfunction'sfn,*args, and**kwargsparameters lack type annotations, and its return typeAny | Noneis unspecified.♻️ Proposed fix
- def wrapper( - completion_config: NativeCompletionConfig, query: QueryParams, **kwargs - ): + def wrapper( + completion_config: NativeCompletionConfig, + query: QueryParams, + **kwargs: Any, + ) -> tuple[LLMCallResponse | None, str | None]: @@ - def langfuse_call(fn, *args, **kwargs): + def langfuse_call( + fn: Callable[..., Any], + *args: Any, + **kwargs: Any, + ) -> Any | None:
♻️ Duplicate comments (2)
backend/app/core/langfuse/langfuse.py (2)
62-72: Add typing + function-name log prefix in_langfuse_call.
This repeats an earlier note:_langfuse_callstill lacks type annotations and the log prefix should include the method name. As per coding guidelines, ...♻️ Proposed fix
- def _langfuse_call(self, fn, *args, **kwargs): + def _langfuse_call( + self, + fn: Callable[..., Any], + *args: Any, + **kwargs: Any, + ) -> Any | None: if self._failed: return None try: return fn(*args, **kwargs) except Exception as e: logger.warning( - f"[LangfuseTracer] {getattr(fn, '__name__', 'operation')} failed: {e}" + f"[LangfuseTracer._langfuse_call] {getattr(fn, '__name__', 'operation')} failed: {e}" ) self._failed = True return None#!/bin/bash # Verify current signature/log prefix for _langfuse_call rg -n "def _langfuse_call" backend/app/core/langfuse/langfuse.py rg -n "LangfuseTracer\._langfuse_call" backend/app/core/langfuse/langfuse.py
168-183: Prefix wrapper logs with the function name.
These log lines should use a function-name prefix (e.g.,[observe_llm_execution.wrapper]/[observe_llm_execution.langfuse_call]) rather than[Langfuse]. As per coding guidelines, ...♻️ Proposed fix
- logger.info("[Langfuse] No credentials - skipping observability") + logger.info( + "[observe_llm_execution.wrapper] No credentials - skipping observability" + ) @@ - logger.info( - f"[Langfuse] Tracing enabled | session_id={session_id or 'auto'}" - ) + logger.info( + f"[observe_llm_execution.wrapper] Tracing enabled | session_id={session_id or 'auto'}" + ) @@ - logger.warning(f"[Langfuse] Failed to initialize client: {e}") + logger.warning( + f"[observe_llm_execution.wrapper] Failed to initialize client: {e}" + ) @@ - logger.warning( - f"[Langfuse] {getattr(fn, '__name__', 'operation')} failed: {e}" - ) + logger.warning( + f"[observe_llm_execution.langfuse_call] {getattr(fn, '__name__', 'operation')} failed: {e}" + )Also applies to: 195-197
🧹 Nitpick comments (1)
backend/app/core/langfuse/langfuse.py (1)
35-44: Log Langfuse init failures for diagnosability.
Silently returning on init failure makes troubleshooting hard; consider logging the exception with the function-name prefix. As per coding guidelines, ...♻️ Proposed tweak
- except Exception: + except Exception as e: + logger.warning( + f"[LangfuseTracer.__init__] Failed to initialize Langfuse: {e}" + ) self._failed = True return
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/app/core/langfuse/langfuse.py (1)
161-197: Add missing type hints forwrapperandlangfuse_call.The
wrapperfunction lacks**kwargs: Anyand return typetuple[LLMCallResponse | None, str | None]. The innerlangfuse_callfunction lacks annotations forfn,*args,**kwargs, and its return typeAny | None. Per coding guidelines, all function parameters and return values must have type hints.♻️ Proposed fix
- def wrapper( - completion_config: NativeCompletionConfig, query: QueryParams, **kwargs - ): + def wrapper( + completion_config: NativeCompletionConfig, + query: QueryParams, + **kwargs: Any, + ) -> tuple[LLMCallResponse | None, str | None]: @@ - def langfuse_call(fn, *args, **kwargs): + def langfuse_call( + fn: Callable[..., Any], + *args: Any, + **kwargs: Any, + ) -> Any | None:
🤖 Fix all issues with AI agents
In `@backend/app/core/langfuse/langfuse.py`:
- Around line 35-60: Update the log messages in LangfuseTracer.__init__ to
follow the logging convention by prefixing them with
"[LangfuseTracer.__init__]": change the logger.warning that logs initialization
failure (inside the except for Langfuse(...)), the logger.debug in the except
when fetch_traces/session resume fails, the logger.info that announces tracing
enabled with session_id, and the logger.warning that reports tracing disabled
due to missing credentials; keep the same messages and exception interpolation
but prepend the required function-name tag so all logs from the constructor use
"[LangfuseTracer.__init__]".
♻️ Duplicate comments (2)
backend/app/core/langfuse/langfuse.py (2)
61-71: Use a function-name prefix inside_langfuse_calllogs.Line 67 should include the method name prefix to align with the logging convention. As per coding guidelines, ...
♻️ Proposed fix
- logger.warning( - f"[LangfuseTracer] {getattr(fn, '__name__', 'operation')} failed: {e}" - ) + logger.warning( + f"[LangfuseTracer._langfuse_call] {getattr(fn, '__name__', 'operation')} failed: {e}" + )
198-200: Use a function-name prefix insidelangfuse_calllogs.Line 198 should reflect the inner function name for consistency with the logging convention. As per coding guidelines, ...
♻️ Proposed fix
- logger.warning( - f"[observe_llm_execution] {getattr(fn, '__name__', 'operation')} failed: {e}" - ) + logger.warning( + f"[observe_llm_execution.langfuse_call] {getattr(fn, '__name__', 'operation')} failed: {e}" + )
| try: | ||
| self.langfuse = Langfuse( | ||
| public_key=credentials["public_key"], | ||
| secret_key=credentials["secret_key"], | ||
| host=credentials["host"], | ||
| enabled=True, # This ensures the client is active | ||
| ) | ||
| except Exception as e: | ||
| logger.warning(f"[LangfuseTracer] Failed to initialize: {e}") | ||
| self._failed = True | ||
| return | ||
|
|
||
| if response_id: | ||
| traces = self.langfuse.fetch_traces(tags=response_id).data | ||
| if traces: | ||
| self.session_id = traces[0].session_id | ||
| try: | ||
| traces = self.langfuse.fetch_traces(tags=response_id).data | ||
| if traces: | ||
| self.session_id = traces[0].session_id | ||
| except Exception as e: | ||
| logger.debug(f"[LangfuseTracer] Session resume failed: {e}") | ||
|
|
||
| logger.info( | ||
| f"[LangfuseTracer] Langfuse tracing enabled | session_id={self.session_id}" | ||
| f"[LangfuseTracer] Tracing enabled | session_id={self.session_id}" | ||
| ) | ||
| else: | ||
| self.langfuse = Langfuse(enabled=False) | ||
| logger.warning("[LangfuseTracer] Tracing disabled - missing credentials") | ||
|
|
There was a problem hiding this comment.
Prefix __init__ logs with the function name.
Lines 43, 53, 55-59 should use [LangfuseTracer.__init__] to match the logging convention. As per coding guidelines, ...
♻️ Proposed fix
- logger.warning(f"[LangfuseTracer] Failed to initialize: {e}")
+ logger.warning(f"[LangfuseTracer.__init__] Failed to initialize: {e}")
@@
- logger.debug(f"[LangfuseTracer] Session resume failed: {e}")
+ logger.debug(f"[LangfuseTracer.__init__] Session resume failed: {e}")
@@
- logger.info(
- f"[LangfuseTracer] Tracing enabled | session_id={self.session_id}"
- )
+ logger.info(
+ f"[LangfuseTracer.__init__] Tracing enabled | session_id={self.session_id}"
+ )
@@
- logger.warning("[LangfuseTracer] Tracing disabled - missing credentials")
+ logger.warning("[LangfuseTracer.__init__] Tracing disabled - missing credentials")🤖 Prompt for AI Agents
In `@backend/app/core/langfuse/langfuse.py` around lines 35 - 60, Update the log
messages in LangfuseTracer.__init__ to follow the logging convention by
prefixing them with "[LangfuseTracer.__init__]": change the logger.warning that
logs initialization failure (inside the except for Langfuse(...)), the
logger.debug in the except when fetch_traces/session resume fails, the
logger.info that announces tracing enabled with session_id, and the
logger.warning that reports tracing disabled due to missing credentials; keep
the same messages and exception interpolation but prepend the required
function-name tag so all logs from the constructor use
"[LangfuseTracer.__init__]".
Fixes issue where Langfuse connection failures could block API responses.
Summary
Target issue is #510
Explain the motivation for making this change. What existing problem does the pull request solve?
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.