LangGraph simple example

Author: benjibc

eval_protocol/adapters/bigquery.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Optional, Union, cast, TypeAlias
+from typing import Any, Callable, Dict, Iterator, List, Optional, TypeAlias
 
 from eval_protocol.models import CompletionParams, EvaluationRow, InputMetadata, Message
 
@@ -108,10 +108,7 @@ def __init__(
             # Avoid strict typing on optional dependency
             self.client = _bigquery_runtime.Client(**client_args)  # type: ignore[no-untyped-call, assignment]
 
-        except DefaultCredentialsError as e:
-            logger.error("Failed to authenticate with BigQuery: %s", e)
-            raise
-        except Exception as e:
+        except (DefaultCredentialsError, ImportError, ValueError, TypeError) as e:
             logger.error("Failed to initialize BigQuery client: %s", e)
             raise
 
@@ -191,10 +188,7 @@ def get_evaluation_rows(
 
                 row_count += 1
 
-        except (NotFound, Forbidden) as e:
-            logger.error("BigQuery access error: %s", e)
-            raise
-        except Exception as e:
+        except (NotFound, Forbidden, RuntimeError, ValueError, TypeError, AttributeError) as e:
             logger.error("Error executing BigQuery query: %s", e)
             raise
 

eval_protocol/adapters/langchain.py

@@ -3,7 +3,7 @@
 import os
 from typing import Any, Dict, List, Optional
 
-from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage, ToolMessage
 
 from eval_protocol.models import Message
 
@@ -49,75 +49,12 @@ def serialize_lc_message_to_ep(msg: BaseMessage) -> Message:
                     parts.append(item)
             content = "\n".join(parts)
 
-        tool_calls_payload: Optional[List[Dict[str, Any]]] = None
-
-        def _normalize_tool_calls(tc_list: List[Any]) -> List[Dict[str, Any]]:
-            mapped: List[Dict[str, Any]] = []
-            for call in tc_list:
-                if not isinstance(call, dict):
-                    continue
-                try:
-                    call_id = call.get("id") or "toolcall_0"
-                    if isinstance(call.get("function"), dict):
-                        fn = call["function"]
-                        fn_name = fn.get("name") or call.get("name") or "tool"
-                        fn_args = fn.get("arguments")
-                    else:
-                        fn_name = call.get("name") or "tool"
-                        fn_args = call.get("arguments") if call.get("arguments") is not None else call.get("args")
-                    if not isinstance(fn_args, str):
-                        import json as _json
-
-                        fn_args = _json.dumps(fn_args or {}, ensure_ascii=False)
-                    mapped.append(
-                        {
-                            "id": call_id,
-                            "type": "function",
-                            "function": {"name": fn_name, "arguments": fn_args},
-                        }
-                    )
-                except Exception:
-                    continue
-            return mapped
-
-        ak = getattr(msg, "additional_kwargs", None)
-        if isinstance(ak, dict):
-            tc = ak.get("tool_calls")
-            if isinstance(tc, list) and tc:
-                mapped = _normalize_tool_calls(tc)
-                if mapped:
-                    tool_calls_payload = mapped
-
-        if tool_calls_payload is None:
-            raw_attr_tc = getattr(msg, "tool_calls", None)
-            if isinstance(raw_attr_tc, list) and raw_attr_tc:
-                mapped = _normalize_tool_calls(raw_attr_tc)
-                if mapped:
-                    tool_calls_payload = mapped
-
-        # Extract reasoning/thinking parts into reasoning_content
-        reasoning_content = None
-        if isinstance(msg.content, list):
-            collected = [
-                it.get("thinking", "") for it in msg.content if isinstance(it, dict) and it.get("type") == "thinking"
-            ]
-            if collected:
-                reasoning_content = "\n\n".join([s for s in collected if s]) or None
-
-        # Message.tool_calls expects List[ChatCompletionMessageToolCall] | None.
-        # We pass through Dicts at runtime but avoid type error by casting.
-        ep_msg = Message(
-            role="assistant",
-            content=content,
-            tool_calls=tool_calls_payload,  # type: ignore[arg-type]
-            reasoning_content=reasoning_content,
-        )
+        ep_msg = Message(role="assistant", content=content)
         _dbg_print(
             "[EP-Ser] -> EP Message:",
             {
                 "role": ep_msg.role,
                 "content_len": len(ep_msg.content or ""),
-                "tool_calls": len(ep_msg.tool_calls or []) if isinstance(ep_msg.tool_calls, list) else 0,
             },
         )
         return ep_msg
@@ -141,3 +78,38 @@ def _normalize_tool_calls(tc_list: List[Any]) -> List[Dict[str, Any]]:
     ep_msg = Message(role=getattr(msg, "type", "assistant"), content=str(getattr(msg, "content", "")))
     _dbg_print("[EP-Ser] -> EP Message (fallback):", {"role": ep_msg.role, "len": len(ep_msg.content or "")})
     return ep_msg
+
+
+def serialize_ep_messages_to_lc(messages: List[Message]) -> List[BaseMessage]:
+    """Convert eval_protocol Message objects to LangChain BaseMessage list.
+
+    - Flattens content parts into strings when content is a list
+    - Maps EP roles to LC message classes
+    """
+    lc_messages: List[BaseMessage] = []
+    for m in messages or []:
+        content = m.content
+        if isinstance(content, list):
+            text_parts: List[str] = []
+            for part in content:
+                try:
+                    text_parts.append(getattr(part, "text", ""))
+                except AttributeError:
+                    pass
+            content = "\n".join([t for t in text_parts if t])
+        if content is None:
+            content = ""
+        text = str(content)
+
+        role = (m.role or "").lower()
+        if role == "user":
+            lc_messages.append(HumanMessage(content=text))
+        elif role == "assistant":
+            lc_messages.append(AIMessage(content=text))
+        elif role == "system":
+            from langchain_core.messages import SystemMessage  # local import to avoid unused import
+
+            lc_messages.append(SystemMessage(content=text))
+        else:
+            lc_messages.append(HumanMessage(content=text))
+    return lc_messages

eval_protocol/pytest/default_langchain_rollout_processor.py

@@ -1,17 +1,25 @@
 import asyncio
 import time
-from typing import List
+from typing import List, Any, cast
 
 try:
-    from langchain_core.messages import BaseMessage
-except Exception:  # pragma: no cover - optional dependency path
-    # Minimal fallback base type to satisfy typing when langchain is not present
-    class BaseMessage:  # type: ignore
-        pass
+    from langchain_core.messages import BaseMessage as LCBaseMessage, HumanMessage  # type: ignore
+except ImportError:  # pragma: no cover - optional dependency path
+    # Minimal fallbacks to satisfy typing when langchain is not present
+    class LCBaseMessage:  # type: ignore
+        content: str
+        type: str
+
+        def __init__(self, content: str = "", msg_type: str = "assistant"):
+            self.content = content
+            self.type = msg_type
+
+    class HumanMessage(LCBaseMessage):  # type: ignore
+        def __init__(self, content: str):
+            super().__init__(content=content, msg_type="human")
 
 
 from eval_protocol.models import EvaluationRow, Message
-from openai.types import CompletionUsage
 from eval_protocol.pytest.rollout_processor import RolloutProcessor
 from eval_protocol.pytest.types import RolloutProcessorConfig
 
@@ -34,27 +42,17 @@ def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig):
         async def _process_row(row: EvaluationRow) -> EvaluationRow:
             start_time = time.perf_counter()
 
-            # Build LC messages from EP row
-            try:
-                from langchain_core.messages import HumanMessage
-            except Exception:
-                # Fallback minimal message if langchain_core is unavailable
-                class HumanMessage(BaseMessage):  # type: ignore
-                    def __init__(self, content: str):
-                        self.content = content
-                        self.type = "human"
-
-            lm_messages: List[BaseMessage] = []
+            # Build LC messages from EP row (minimal: last user to HumanMessage)
+            lm_messages: List[LCBaseMessage] = []
             if row.messages:
                 last_user = [m for m in row.messages if m.role == "user"]
                 if last_user:
                     content = last_user[-1].content or ""
                     if isinstance(content, list):
-                        # Flatten our SDK content parts into a single string for LangChain
                         content = "".join([getattr(p, "text", str(p)) for p in content])
                     lm_messages.append(HumanMessage(content=str(content)))
             if not lm_messages:
-                lm_messages = [HumanMessage(content="")]  # minimal
+                lm_messages = [HumanMessage(content="")]
 
             target = await self.get_invoke_target(config)
 
@@ -72,7 +70,7 @@ async def _invoke_direct(payload):
 
                 invoke_fn = _invoke_direct
             elif callable(target):
-                # If target is a normal callable, call it directly; if it returns an awaitable, await it
+
                 async def _invoke_wrapper(payload):
                     result = target(payload)
                     if asyncio.iscoroutine(result):
@@ -84,44 +82,18 @@ async def _invoke_wrapper(payload):
                 raise TypeError("Unsupported invoke target for LangGraphRolloutProcessor")
 
             result_obj = await invoke_fn({"messages": lm_messages})
-            # Accept both dicts and objects with .get/.messages
             if isinstance(result_obj, dict):
-                result_messages: List[BaseMessage] = result_obj.get("messages", [])
+                result_messages: List[LCBaseMessage] = result_obj.get("messages", [])
             else:
                 result_messages = getattr(result_obj, "messages", [])
 
-            # TODO: i didn't see a langgraph example so couldn't fully test this. should uncomment and test when we have example ready.
-            # total_input_tokens = 0
-            # total_output_tokens = 0
-            # total_tokens = 0
-
-            # for msg in result_messages:
-            #     if isinstance(msg, BaseMessage):
-            #         usage = getattr(msg, 'response_metadata', {})
-            #     else:
-            #         usage = msg.get("response_metadata", {})
-
-            #     if usage:
-            #         total_input_tokens += usage.get("prompt_tokens", 0)
-            #         total_output_tokens += usage.get("completion_tokens", 0)
-            #         total_tokens += usage.get("total_tokens", 0)
-
-            # row.execution_metadata.usage = CompletionUsage(
-            #     prompt_tokens=total_input_tokens,
-            #     completion_tokens=total_output_tokens,
-            #     total_tokens=total_tokens,
-            # )
-
-            def _serialize_message(msg: BaseMessage) -> Message:
-                # Prefer SDK-level serializer
+            def _serialize_message(msg: LCBaseMessage) -> Message:
                 try:
                     from eval_protocol.adapters.langchain import serialize_lc_message_to_ep as _ser
-
-                    return _ser(msg)
-                except Exception:
-                    # Minimal fallback: best-effort string content only
+                except ImportError:
                     content = getattr(msg, "content", "")
                     return Message(role=getattr(msg, "type", "assistant"), content=str(content))
+                return _ser(cast(Any, msg))
 
             row.messages = [_serialize_message(m) for m in result_messages]
 

eval_protocol/pytest/handle_persist_flow.py

@@ -42,6 +42,13 @@ def handle_persist_flow(all_results: list[list[EvaluationRow]], test_func_name:
                     if len(dataset_name) > 63:
                         dataset_name = dataset_name[:63]
 
+                    # Fireworks requires: last character of id must not be '-'
+                    dataset_name = dataset_name.rstrip("-")
+
+                    # Ensure non-empty after stripping; fallback to safe_test_func_name
+                    if not dataset_name:
+                        dataset_name = safe_test_func_name[:63].rstrip("-") or "dataset"
+
                     exp_file = exp_dir / f"{experiment_id}.jsonl"
                     with open(exp_file, "w", encoding="utf-8") as f:
                         for row in exp_rows: