Добавлены переменные для langfuse

.env.example

@@ -3,3 +3,6 @@ PROVIDER_API_KEY=your-api-key
 PROVIDER_MODEL=gpt-4
 COMPOSIO_API_KEY=your-api-key
 AGENT_ID=user-12345
+LANGFUSE_PUBLIC_KEY=pk-lf-...
+LANGFUSE_SECRET_KEY=sk-lf-...
+LANGFUSE_HOST=http://localhost:3000

Dockerfile

@@ -29,7 +29,6 @@ FROM base AS production
 
 COPY --from=builder /app/.venv /app/.venv
 ENV PATH="/app/.venv/bin:$PATH"
-ENV ENVIRONMENT="prod"
 
 COPY src/ /app/src/
 COPY configs/ /app/configs/
@@ -54,7 +53,6 @@ COPY --from=agent_api . /agent_api/
 RUN uv pip install -e /agent_api/
 
 ENV PATH="/app/.venv/bin:$PATH"
-ENV ENVIRONMENT="dev"
 
 COPY Makefile ./
 COPY .mk/ ./.mk/

configs/logging_test.yaml

@@ -0,0 +1,29 @@
+version: 1
+disable_existing_loggers: false
+
+formatters:
+  test_formatter:
+    format: "%(asctime)s | %(correlation_id)s | %(levelname)-7s | %(name)s | %(message)s"
+    datefmt: "%Y-%m-%d %H:%M:%S"
+
+handlers:
+  console:
+    class: logging.StreamHandler
+    level: DEBUG
+    formatter: test_formatter
+    stream: ext://sys.stdout
+    filters: [correlation_filter]
+
+filters:
+  correlation_filter:
+    (): src.core.correlation.CorrelationFilter
+
+loggers:
+  src:
+    level: DEBUG
+    handlers: [console]
+    propagate: false
+
+root:
+  level: WARNING
+  handlers: [console]

src/agent/service.py

@@ -2,7 +2,9 @@ from typing import AsyncIterator, AsyncContextManager, Self
 from abc import abstractmethod
 import os
 from pathlib import Path
-
+from langfuse import get_client, propagate_attributes,observe
+import langfuse
+from langfuse.langchain import CallbackHandler
 from src.agent.base import create_agent
 from src.core.logger import get_logger
 from lambda_agent_api.server import (
@@ -106,8 +108,18 @@ class AgentService:
     async def __astream(
         self, chat_id: int, text: str, attachments: list[str] = None
     ) -> AsyncIterator[AgentEventUnion]:
-        config = {"configurable": {"thread_id": chat_id}}
         
+        langfuse_client = get_client()
+        langfuse_handler = CallbackHandler()
+        config = {"configurable": {"thread_id": chat_id},"callbacks": [langfuse_handler]}
+
+        try:
+            # возможно сюда стоит включить обработку attachmanets, только  __describe_attachments нужно обернуть в @observe
+            with propagate_attributes(
+                session_id=str(chat_id),
+                user_id=str(chat_id),   # Или реальный ID юзера, если есть
+                tags=["agent-stream"]
+            ):
                 new_message = text
                 if attachments:
                     logger.debug(f"Processing {attachments} attachments for chat {chat_id}")
@@ -115,13 +127,13 @@ class AgentService:
                     new_message += "\n" + attachments_description
                     
                 logger.debug(f"Starting agent stream for chat {chat_id}")
-        
                 # astream_events, чтобы выходили промежуточные ивенты по мере их генерации
                 async for event in self._agent.astream_events(
                     {"messages": [{"role": "user", "content": new_message}]},
                     config=config,
                     version="v2",
                 ):
+        
                     kind = event["event"]
 
                     # пришли чанки от LLM
@@ -162,7 +174,11 @@ class AgentService:
                             yield MsgEventSendFile(path=file_path)
                 
                 logger.debug(f"Agent stream completed for chat {chat_id}")
+        finally:
+            logger.debug(f"Flushing Langfuse logs for chat {chat_id}")
+            langfuse_client.flush()
 
+    @observe(as_type="span", name="parse_attachments")
     async def __describe_attachments(self, raw_paths: list[str]) -> str:
         lines = []
         logger.debug(f"Describing {len(raw_paths)} attachments")

src/agent/tools/execute_shell.py

@@ -66,12 +66,12 @@ def execute_shell(
         output = result.stdout
 
         if output:
-            logger.trace(f"Command output STDOUT: {output}")
+            logger.trace(f"Command output: {len(output)} characters on stdout")
 
         if result.stderr:
+            logger.debug("Command output errors on stderr")
             stderr_lines = result.stderr.strip().split("\n")
             output += "\n" + "\n".join(f"[stderr] {line}" for line in stderr_lines)
-            logger.debug(f"Command output STDERR: {result.stderr}")
 
         # Truncate if needed
         max_output = DEFAULT_MAX_OUTPUT