diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ All notable changes to this package will be documented in this file. +## 0.0.6.2 - 2026-11-14 + +> **Note**: `0.0.6.1` was skipped globally after `langchain-hs-ollama-v0.0.6.1`. + +- **Tool-Call & Agent Streaming**: + - Extended `StreamEvent` in `Langchain.Stream` with `ToolStart`, `ToolChunk`, `ToolEnd`, and `ToolErrorEvent`, plus `liftChatStream` and `tapStreamMessage`. + - Added cancellation-safe streaming combinators `liftChatStreamResource` and `liftChatStreamWithCancel` in `Langchain.Stream`. + - Added `StreamingTool`, `createStreamingTool`, `fromTool`, `streamToolCall`, and `streamStreamingToolCall` to `Langchain.Tool`. + - Added `streamingShellTool` (`Langchain.Tool.Shell`) and `withStreamingSecurity` (`Langchain.Tool.Security`). + - Added `withStreamingTools`, `reactStepStream`, `streamReActAgent`, `streamReActAgentWithTrace`, and `runReActAgentStream` to `Langchain.Agent.ReAct`. + - Added `streamPlanAndExecute` and `runPlanAndExecuteStream` to `Langchain.Agent.PlanAndExecute`. + - Updated `Langchain.Prelude` exports for streaming tools and agents. +- **Multi-Agent Primitives & ReAct Loop Detection**: + - Added `LoopDetectionStrategy` (`NoLoopDetection`, `WarnOnDuplicateAction`, `FailOnLoop`, `BreakOnLoop`) and iteration telemetry (`traceIterations`, `traceTerminatedBy`). + - Added multi-agent handoff tools `asAgentTool` and `asStreamingAgentTool`. +- **Evaluation Harness (`Langchain.Eval`)**: + - Added `EvalCase`, `EvalScore`, `Scorer`, and `EvalReport` with Aeson support. + - Implemented `exactMatchScorer`, `containsScorer`, `toolCallExactScorer`, `llmJudgeScorer`, and `runEvalSuite`. +- **Security, Checkpointing, Memory & Parsers (consolidated post-`0.0.6.0` alignment)**: + - `SecurityPolicy`, `FileSystem`, `Process`, and `Checkpoint` tool suites (`Langchain.Tool.*`). + - `InVariableMemory`, `SessionMemory`, `EntityMemory`, `CombinedMemory`, `FileMemory`, and buffer reset helpers. + - Updated `sqlite-simple` bounds to `>= 0.4.18 && < 0.5`. + ## 0.0.6.0 - 2026-10-02 - Updated bounds for GHC 9.12 and modern Stackage LTS releases. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ ```cabal build-depends: - langchain-hs-core >= 0.0.6.0 && < 0.0.7.0 + langchain-hs-core >= 0.0.6.2 && < 0.0.7.0 ``` To connect concrete LLMs, combine with satellite provider packages: diff --git a/langchain-hs-core.cabal b/langchain-hs-core.cabal --- a/langchain-hs-core.cabal +++ b/langchain-hs-core.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: langchain-hs-core -version: 0.0.6.0 +version: 0.0.6.2 synopsis: Core typeclasses and pure AST primitives for langchain-hs description: Zero HTTP dependency core package containing ChatModel, RunnableTree, ContentBlock, Tool, and StreamEvent primitives. category: Web, AI, Control @@ -91,6 +91,7 @@ Langchain.Agent.PlanAndExecute Langchain.Guardrail Langchain.Guardrail.Core + Langchain.Eval Langchain.Prelude other-modules: Paths_langchain_hs_core @@ -127,7 +128,7 @@ , sqlite-simple >=0.4.18 && <0.5 , stm ==2.5.* , text >=1.2 && <3 - , time >=1.9 && <1.15 + , time >=1.9 && <1.16 default-language: Haskell2010 test-suite langchain-hs-core-test @@ -135,6 +136,7 @@ main-is: Spec.hs other-modules: Test.Langchain.Callback + Test.Langchain.Eval Test.Langchain.Model Test.Langchain.Monad Test.Langchain.Observability @@ -185,5 +187,5 @@ , tasty-hunit >=0.10 , tasty-quickcheck >=0.10 , text >=1.2 && <3 - , time >=1.9 && <1.15 + , time >=1.9 && <1.16 default-language: Haskell2010 diff --git a/src/Langchain/Agent/PlanAndExecute.hs b/src/Langchain/Agent/PlanAndExecute.hs --- a/src/Langchain/Agent/PlanAndExecute.hs +++ b/src/Langchain/Agent/PlanAndExecute.hs @@ -27,22 +27,36 @@ ( PlanStep (..) , Plan (..) , StepExecutor (..) + , StreamingStepExecutor (..) , PlanAndExecuteAgent (..) , newPlanAndExecuteAgent , newPlanAndExecuteAgentWithTools , runPlanAndExecute + , streamPlanAndExecute + , streamPlanAndExecuteWithResult + , runPlanAndExecuteStream ) where import Control.Applicative ((<|>)) +import Control.Monad (void) import Control.Monad.Except (MonadError, throwError) import Control.Monad.IO.Class (MonadIO) -import Data.Aeson (FromJSON (..), ToJSON, Value (..), withObject, (.!=), (.:), (.:?)) +import Control.Monad.Trans (lift) +import Data.Aeson (FromJSON (..), ToJSON (toJSON), Value (..), withObject, (.!=), (.:), (.:?)) import Data.Aeson.Types (prependFailure, typeMismatch) +import Data.Conduit (ConduitT, fuseUpstream, runConduit, yield, (.|)) +import qualified Data.Conduit.List as CL import Data.Text (Text) import qualified Data.Text as T import GHC.Generics (Generic) -import Langchain.Agent.ReAct (ReActAgent, defaultReActAgent, runReActAgent) +import Langchain.Agent.ReAct + ( AgentTrace (traceFinalAnswer) + , ReActAgent + , defaultReActAgent + , runReActAgent + , streamReActAgentWithTrace + ) import Langchain.Error (LangchainError, agentError) import Langchain.Model ( ChatModel (..) @@ -50,6 +64,7 @@ , userMessage ) import Langchain.OutputParser.Structured (StructuredOutput, TypeSchema, structuredInvoke) +import Langchain.Stream (EventStream, StreamEvent (..), liftChatStream, tapStreamMessage) import Langchain.Tool (Tool) import Langchain.Tool.Binding (ToolBinder (..)) @@ -129,6 +144,45 @@ msg <- invoke model [userMessage prompt] Nothing pure $ extractMessageText msg +-- | Streaming abstraction for executing individual steps of a plan. +class StreamingStepExecutor e m where + executeStepStream :: + -- | Step executor instance (agent, model with tools, function, or model) + e -> + -- | Step instruction prompt to execute + Text -> + -- | Stream of execution events returning the textual step result + ConduitT () StreamEvent m Text + +instance + {-# OVERLAPPING #-} + (ToolBinder model n, MonadIO n, MonadError LangchainError n) => + StreamingStepExecutor (ReActAgent model n) n + where + executeStepStream agent prompt = do + trace <- streamReActAgentWithTrace agent [userMessage prompt] + pure $ extractMessageText (traceFinalAnswer trace) + +instance + {-# OVERLAPPING #-} + (ToolBinder model n, MonadIO n, MonadError LangchainError n) => + StreamingStepExecutor (model, [Tool n]) n + where + executeStepStream (model, tools) = + executeStepStream (defaultReActAgent model tools) + +instance {-# OVERLAPPING #-} (m ~ n, Monad m) => StreamingStepExecutor (Text -> m Text) n where + executeStepStream f prompt = lift (f prompt) + +instance + {-# OVERLAPPABLE #-} + (ChatModel model, MonadIO m, MonadError LangchainError m) => + StreamingStepExecutor model m + where + executeStepStream model prompt = do + msg <- liftChatStream (stream model [userMessage prompt] Nothing) .| tapStreamMessage + pure $ extractMessageText msg + -- | Plan-and-Execute agent container data PlanAndExecuteAgent planner executor = PlanAndExecuteAgent { plannerModel :: planner @@ -163,6 +217,41 @@ newPlanAndExecuteAgentWithTools planner model tools = PlanAndExecuteAgent planner (defaultReActAgent model tools) +buildPlanPrompt :: Maybe Text -> Text -> Text +buildPlanPrompt (Just p) userGoal = p <> "\nGoal: " <> userGoal +buildPlanPrompt Nothing userGoal = + "You are an expert planner. For the following goal, generate a concise step-by-step execution plan.\n" + <> "Output JSON format: {\"planSteps\": [{\"stepNumber\": 1, \"stepDescription\": \"...\"}]}\n" + <> "Keep the plan focused and minimal (between 2 to 3 distinct, actionable steps).\n" + <> "Goal: " + <> userGoal + +buildStepPrompt :: Text -> [(PlanStep, Text)] -> PlanStep -> Text +buildStepPrompt userGoal prevOutputs currStep = + "User Goal: " + <> userGoal + <> ( if null prevOutputs + then "" + else + "\n\nCompleted Steps So Far:\n" + <> T.unlines + [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- prevOutputs] + ) + <> "\n\nCurrent Task To Execute (Step " + <> T.pack (show (stepNumber currStep)) + <> "): " + <> stepDescription currStep + <> "\nExecute this task using any appropriate tools available and provide the outcome:" + +buildSynthesisPrompt :: Text -> [(PlanStep, Text)] -> Text +buildSynthesisPrompt userGoal stepOutputs = + "User Goal: " + <> userGoal + <> "\n\nStep Execution History:\n" + <> T.unlines + [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- stepOutputs] + <> "\n\nProvide the final synthesized answer satisfying the goal:" + -- | Execute a goal using the Plan-and-Execute workflow with structured JSON planning runPlanAndExecute :: (ChatModel planner, StepExecutor executor m, MonadIO m, MonadError LangchainError m) => @@ -173,43 +262,60 @@ -- | Final synthesized answer solving the user goal m Text runPlanAndExecute PlanAndExecuteAgent {..} userGoal = do - let planPrompt = case planPromptTemplate of - Just p -> p <> "\nGoal: " <> userGoal - Nothing -> - "You are an expert planner. For the following goal, generate a concise step-by-step execution plan.\n" - <> "Output JSON format: {\"planSteps\": [{\"stepNumber\": 1, \"stepDescription\": \"...\"}]}\n" - <> "Keep the plan focused and minimal (between 2 to 3 distinct, actionable steps).\n" - <> "Goal: " - <> userGoal - plan <- structuredInvoke plannerModel [userMessage planPrompt] + plan <- structuredInvoke plannerModel [userMessage (buildPlanPrompt planPromptTemplate userGoal)] if null (planSteps plan) then throwError $ agentError "Planner generated an empty plan" (Just "PlanAndExecuteAgent") Nothing else executeSteps (planSteps plan) [] where - executeSteps [] stepOutputs = do - let synthesisPrompt = - "User Goal: " - <> userGoal - <> "\n\nStep Execution History:\n" - <> T.unlines - [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- stepOutputs] - <> "\n\nProvide the final synthesized answer satisfying the goal:" - executeStep stepExecutor synthesisPrompt + executeSteps [] stepOutputs = + executeStep stepExecutor (buildSynthesisPrompt userGoal stepOutputs) executeSteps (currStep : restSteps) prevOutputs = do - let stepPrompt = - "User Goal: " - <> userGoal - <> ( if null prevOutputs - then "" - else - "\n\nCompleted Steps So Far:\n" - <> T.unlines - [T.pack (show num) <> ". " <> desc <> " -> " <> out | (PlanStep num desc, out) <- prevOutputs] - ) - <> "\n\nCurrent Task To Execute (Step " - <> T.pack (show (stepNumber currStep)) - <> "): " - <> stepDescription currStep - <> "\nExecute this task using any appropriate tools available and provide the outcome:" - stepOut <- executeStep stepExecutor stepPrompt + stepOut <- executeStep stepExecutor (buildStepPrompt userGoal prevOutputs currStep) executeSteps restSteps (prevOutputs ++ [(currStep, stepOut)]) + +{- | Execute a goal using the Plan-and-Execute workflow while streaming 'StreamEvent's +and returning the final synthesized answer. +-} +streamPlanAndExecuteWithResult :: + (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) => + PlanAndExecuteAgent planner executor -> + Text -> + ConduitT () StreamEvent m Text +streamPlanAndExecuteWithResult PlanAndExecuteAgent {..} userGoal = do + let pRunId = "plan-and-execute-run" + yield $ ChainStart pRunId "PlanAndExecuteAgent" (toJSON userGoal) + plan <- + lift $ structuredInvoke plannerModel [userMessage (buildPlanPrompt planPromptTemplate userGoal)] + if null (planSteps plan) + then throwError $ agentError "Planner generated an empty plan" (Just "PlanAndExecuteAgent") Nothing + else do + finalOut <- executeSteps pRunId (planSteps plan) [] + yield $ ChainEnd pRunId "PlanAndExecuteAgent" (toJSON finalOut) + pure finalOut + where + executeSteps _ [] stepOutputs = + executeStepStream stepExecutor (buildSynthesisPrompt userGoal stepOutputs) + executeSteps pRunId (currStep : restSteps) prevOutputs = do + let sNodeId = "step-" <> T.pack (show (stepNumber currStep)) + yield $ NodeStart pRunId sNodeId (toJSON currStep) + stepOut <- executeStepStream stepExecutor (buildStepPrompt userGoal prevOutputs currStep) + yield $ NodeEnd pRunId sNodeId (toJSON stepOut) + executeSteps pRunId restSteps (prevOutputs ++ [(currStep, stepOut)]) + +-- | Execute a goal using the Plan-and-Execute workflow as an 'EventStream'. +streamPlanAndExecute :: + (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) => + PlanAndExecuteAgent planner executor -> + Text -> + EventStream m +streamPlanAndExecute agent userGoal = void (streamPlanAndExecuteWithResult agent userGoal) + +-- | Execute a Plan-and-Execute agent in streaming mode with an event callback. +runPlanAndExecuteStream :: + (ChatModel planner, StreamingStepExecutor executor m, MonadIO m, MonadError LangchainError m) => + PlanAndExecuteAgent planner executor -> + Text -> + (StreamEvent -> m ()) -> + m Text +runPlanAndExecuteStream agent userGoal onEvent = + runConduit (streamPlanAndExecuteWithResult agent userGoal `fuseUpstream` CL.mapM_ onEvent) diff --git a/src/Langchain/Agent/ReAct.hs b/src/Langchain/Agent/ReAct.hs --- a/src/Langchain/Agent/ReAct.hs +++ b/src/Langchain/Agent/ReAct.hs @@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -49,6 +50,7 @@ , AgentTrace (..) , ReActAgent (..) , ToolErrorStrategy (..) + , LoopDetectionStrategy (..) -- * Construction , defaultReActAgent @@ -62,26 +64,67 @@ , withCallbackManager , withStopCondition , withToolErrorStrategy + , withStreamingTools + , withLoopDetection + , withMaxConsecutiveIdenticalToolCalls + , withMaxDelegationDepth + -- * Multi-Agent Handoff & Delegation Tools + , asAgentTool + , asStreamingAgentTool + -- * Execution , reactStep , runReActAgent , runReActAgentWithTrace + + -- * Streaming Execution + , InterruptibleResult (..) + , reactStepStream + , streamReActAgent + , streamReActAgentWithTrace + , runReActAgentStream + , runReActAgentStreamInterruptible ) where -import Control.Monad (forM, when) +import Control.Monad (unless, when) import Control.Monad.Except (MonadError, catchError, throwError) import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Trans (lift) +import Data.Aeson (Value (..), object, toJSON, (.=)) +import Data.Aeson.Types (parseEither, (.:?)) +import Data.Conduit (ConduitT, await, fuseUpstream, mapOutput, runConduit, yield, (.|)) +import qualified Data.Conduit.List as CL +import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef) import Data.List (find) import Data.Text (Text) import qualified Data.Text as T import Data.Time.Clock (diffUTCTime, getCurrentTime) import Langchain.Callback.Manager (CallbackEvent (..), CallbackManager, dispatchEvent) -import Langchain.Error (LangchainError, agentError, errorMessage) +import Langchain.Error (LangchainError, agentError, errorMessage, toolError) import Langchain.Memory.Core (BaseMemory (..), SomeMemory (..)) import Langchain.Model import qualified Langchain.Model.Types as M +import Langchain.Stream + ( CancelToken + , EventStream + , StreamEvent + ( ChainEnd + , ChainStart + , LLMChunk + , LLMEnd + , LLMStart + , ToolChunk + , ToolEnd + , ToolErrorEvent + , ToolStart + ) + , isCancelled + , liftChatStream + , liftChatStreamWithCancel + , tapStreamMessage + ) import Langchain.Tool import Langchain.Tool.Binding (ToolBinder (..)) @@ -93,6 +136,16 @@ FailOnError deriving (Eq, Show) +-- | Strategy for detecting runaway loops where the model repeatedly invokes identical tools with identical arguments. +data LoopDetectionStrategy + = -- | Allow consecutive repeat tool calls without intervention (default for backwards compatibility). + AllowRepeats + | -- | After @n@ consecutive identical tool call signatures, skip tool execution and inject an observation warning to the LLM. + WarnOnRepeat !Int + | -- | After @n@ consecutive identical tool call signatures, abort execution immediately with an 'AgentError' and notify 'CallbackManager'. + AbortOnRepeat !Int + deriving (Eq, Show) + -- | A single step in the ReAct reasoning chain. data AgentStep = -- | The model decided to call one or more tools. @@ -107,6 +160,8 @@ -- ^ Each pair is (the step, observation messages produced) , traceFinalAnswer :: Message -- ^ The final answer message + , traceIterations :: !Int + -- ^ Total reasoning iterations executed } deriving (Eq, Show) @@ -135,6 +190,12 @@ -- ^ Optional early-stop predicate checked after each LLM response , agentToolErrorStrategy :: ToolErrorStrategy -- ^ How to handle tool execution errors (default: 'ContinueOnError') + , agentStreamingTools :: [StreamingTool m] + -- ^ Optional streaming tools capable of emitting incremental 'ToolChunk' events + , agentLoopDetection :: !LoopDetectionStrategy + -- ^ Strategy for runaway consecutive duplicate tool-call loop detection (default: 'AllowRepeats') + , agentMaxDelegationDepth :: !Int + -- ^ Maximum allowed delegation depth for nested sub-agents (default: 5) } {- | Construct a ReAct agent with sensible defaults. @@ -159,6 +220,9 @@ , agentCallbackManager = Nothing , agentStopCondition = Nothing , agentToolErrorStrategy = ContinueOnError + , agentStreamingTools = [] + , agentLoopDetection = AllowRepeats + , agentMaxDelegationDepth = 5 } {- | Attach a 'BaseMemory' instance for persistent conversation history. @@ -263,6 +327,49 @@ ReActAgent model m withToolErrorStrategy strategy agent = agent {agentToolErrorStrategy = strategy} +-- | Attach 'StreamingTool's that emit incremental 'ToolChunk' events during streaming execution. +withStreamingTools :: + [StreamingTool m] -> + ReActAgent model m -> + ReActAgent model m +withStreamingTools sTools agent = agent {agentStreamingTools = sTools} + +{- | Set the runaway loop detection strategy (default: 'AllowRepeats'). + +* 'AllowRepeats': no intervention on repeated identical tool calls. +* 'WarnOnRepeat': skip execution and inject a warning observation once consecutive duplicates reach @n@. +* 'AbortOnRepeat': immediately terminate the reasoning loop and throw 'AgentError' once consecutive duplicates reach @n@. +-} +withLoopDetection :: + LoopDetectionStrategy -> + ReActAgent model m -> + ReActAgent model m +withLoopDetection strat agent = agent {agentLoopDetection = strat} + +-- | Set runaway loop detection to abort after @n@ consecutive duplicate tool calls. +withMaxConsecutiveIdenticalToolCalls :: + Int -> + ReActAgent model m -> + ReActAgent model m +withMaxConsecutiveIdenticalToolCalls n = withLoopDetection (AbortOnRepeat n) + +-- | Set the maximum allowed agent delegation depth (default: 5). +withMaxDelegationDepth :: + Int -> + ReActAgent model m -> + ReActAgent model m +withMaxDelegationDepth d agent = agent {agentMaxDelegationDepth = d} + +-- | Classify a model response 'Message' as 'AgentAction' or 'AgentFinish'. +classifyStepResponse :: Message -> AgentStep +classifyStepResponse responseMsg = + case messageToolCalls responseMsg of + Just tcs@(_ : _) -> AgentAction responseMsg tcs + _ -> AgentFinish responseMsg + +allAgentTools :: ReActAgent model m -> [Tool m] +allAgentTools ReActAgent {..} = agentTools ++ map streamingToolDef agentStreamingTools + {- | Run a single step of ReAct reasoning. Sends the current message history (with tools bound) to the LLM and @@ -285,51 +392,51 @@ reactStep model tools history userCfg = do let toolCfg = bindToolsConfig @model tools userCfg responseMsg <- invoke model history toolCfg - case messageToolCalls responseMsg of - Just tcs@(_ : _) -> pure $ AgentAction responseMsg tcs - _ -> pure $ AgentFinish responseMsg - -{- | Execute the full ReAct reasoning loop, returning only the final answer. - -For the full reasoning trace, use 'runReActAgentWithTrace'. - -When the agent has a 'BaseMemory' attached: + pure $ classifyStepResponse responseMsg -1. Existing conversation history is loaded from memory. -2. The user's input messages are appended. -3. After the loop finishes, both the user input and final answer are saved to memory. --} -runReActAgent :: +reactStepStreamWith :: forall model m. (ToolBinder model m, MonadIO m, MonadError LangchainError m) => - -- | Configured ReAct agent instance - ReActAgent model m -> - -- | Input user messages for this turn + Maybe CancelToken -> + model -> + [Tool m] -> [Message] -> - -- | The final assistant answer message - m Message -runReActAgent agent userMsgs = traceFinalAnswer <$> runReActAgentWithTrace agent userMsgs - -{- | Execute the full ReAct reasoning loop, returning both the final answer -and the complete intermediate reasoning trace. + Maybe (ModelConfig model) -> + ConduitT () StreamEvent m AgentStep +reactStepStreamWith mbTok model tools history userCfg = do + let toolCfg = bindToolsConfig @model tools userCfg + liftFn = maybe liftChatStream liftChatStreamWithCancel mbTok + responseMsg <- liftFn (stream model history toolCfg) .| tapStreamMessage + pure $ classifyStepResponse responseMsg -Tool calls within a single step are executed in sequence within the @m@ monad, -and the observations are fed back to the LLM for the next reasoning step. +{- | Run a single step of ReAct reasoning using streaming LLM invocation. -The trace includes every 'AgentStep' and the observation messages produced -at each iteration, which is invaluable for debugging and observability. +Forwards all 'StreamEvent's from the underlying 'ChatModel' 'stream' +(including 'LLMStart', 'LLMChunk' with text and 'toolCallDelta', and 'LLMEnd') +downstream in real time, and returns the resulting 'AgentStep'. -} -runReActAgentWithTrace :: +reactStepStream :: forall model m. (ToolBinder model m, MonadIO m, MonadError LangchainError m) => - -- | Configured ReAct agent instance + -- | The chat model to stream + model -> + -- | Available tools to bind to the model + [Tool m] -> + -- | Current conversation message history + [Message] -> + -- | Optional provider-specific model configuration + Maybe (ModelConfig model) -> + -- | Stream of LLM events returning the step classification + ConduitT () StreamEvent m AgentStep +reactStepStream = reactStepStreamWith Nothing + +buildInitialHistory :: + (MonadIO m, MonadError LangchainError m) => ReActAgent model m -> - -- | Input user messages for this turn [Message] -> - -- | Full reasoning trace containing all intermediate steps and final answer - m AgentTrace -runReActAgentWithTrace ReActAgent {..} userMsgs = do - -- 1. Build initial history: system prompt + memory + user messages + Text -> + m [Message] +buildInitialHistory ReActAgent {..} userMsgs caller = do memoryMsgs <- case agentMemory of Nothing -> pure [] Just (SomeMemory mem) -> messages mem @@ -347,135 +454,541 @@ agentError "Cannot invoke ReAct agent with empty message history (no user query, memory, or system prompt)" (Just "ReActAgent") - (Just "runReActAgentWithTrace") + (Just caller) - -- 2. Run the reasoning loop - trace <- go initialHistory agentMaxIterations [] + pure initialHistory - -- 3. Persist to memory if available +persistTurnMemory :: + (MonadIO m, MonadError LangchainError m) => + ReActAgent model m -> + [Message] -> + Message -> + m () +persistTurnMemory ReActAgent {..} userMsgs finalMsg = case agentMemory of Nothing -> pure () Just (SomeMemory mem) -> do - -- Save each user message mapM_ (addMessage mem) userMsgs - -- Save the final answer - addMessage mem (traceFinalAnswer trace) + addMessage mem finalMsg - pure trace +toolObservationMessage :: Text -> Text -> Text -> Message +toolObservationMessage tName tcId txt = + (textMessage M.Tool txt) + { M.messageName = Just tName + , M.messageToolId = Just tcId + } + +dispatchCb :: MonadIO m => Maybe CallbackManager -> CallbackEvent -> m () +dispatchCb Nothing _ = pure () +dispatchCb (Just mgr) ev = dispatchEvent mgr ev + +reactRunId :: Text +reactRunId = "react-agent-run" + +-- | Execute a ReAct loop step-by-step, parameterized over the LLM step and tool executors. +runLoopWith :: + forall model m n. + (MonadIO n, MonadError LangchainError n) => + Maybe CancelToken -> + ReActAgent model m -> + ([Message] -> n AgentStep) -> + (ToolCall -> n Message) -> + [Message] -> + Int -> + [(AgentStep, [Message])] -> + n AgentTrace +runLoopWith mbTok agent@ReActAgent {..} runStep execTool = go Nothing 0 where - go :: [Message] -> Int -> [(AgentStep, [Message])] -> m AgentTrace - go history maxIter accSteps + checkCancelled = maybe (pure False) isCancelled mbTok + + execToolsWithCancel [] acc = pure (reverse acc) + execToolsWithCancel (tc : rest) acc = do + cancelled <- checkCancelled + if cancelled + then pure (reverse acc) + else do + obs <- execTool tc + execToolsWithCancel rest (obs : acc) + + toolSignature tcs = [(toolCallName tc, toolCallArguments tc) | tc <- tcs] + + go mbLastSig repeatCount history maxIter accSteps | maxIter <= 0 = do let maxErr = "ReAct Agent exceeded maximum iterations (" <> T.pack (show agentMaxIterations) <> ")" - case agentCallbackManager of - Nothing -> pure () - Just mgr -> do - now <- liftIO getCurrentTime - dispatchEvent mgr (OnError "ReActAgent" maxErr now) + now <- liftIO getCurrentTime + dispatchCb agentCallbackManager (OnError "ReActAgent" maxErr now) throwError $ agentError maxErr (Just "ReActAgent") (Just "runReActAgent") | otherwise = do - -- Dispatch OnLLMStart callback before calling model - llmStartTime <- liftIO getCurrentTime - case agentCallbackManager of - Nothing -> pure () - Just mgr -> - dispatchEvent mgr (OnLLMStart "ReActAgent" (map extractMessageText history) llmStartTime) + cancelledBefore <- checkCancelled + if cancelledBefore + then pure $ AgentTrace accSteps (assistantMessage "") (length accSteps) + else do + llmStartTime <- liftIO getCurrentTime + dispatchCb + agentCallbackManager + (OnLLMStart "ReActAgent" (map extractMessageText history) llmStartTime) - -- Invoke the model with error interception for observability - eStep <- - (Right <$> reactStep agentModel agentTools history agentModelConfig) - `catchError` (pure . Left) - llmEndTime <- liftIO getCurrentTime - let llmLatency = round (realToFrac (diffUTCTime llmEndTime llmStartTime) * 1000000 :: Double) :: Int - case eStep of - Left err -> do - case agentCallbackManager of - Nothing -> pure () - Just mgr -> - dispatchEvent mgr (OnError "ReActAgent" (errorMessage err) llmEndTime) - throwError err - Right step -> case step of - AgentFinish finalMsg -> do - case agentCallbackManager of - Nothing -> pure () - Just mgr -> - dispatchEvent mgr (OnLLMEnd "ReActAgent" (extractMessageText finalMsg) llmLatency llmEndTime) - let allSteps = accSteps ++ [(step, [])] - pure $ AgentTrace allSteps finalMsg - AgentAction respMsg tcs -> do - case agentCallbackManager of - Nothing -> pure () - Just mgr -> - dispatchEvent mgr (OnLLMEnd "ReActAgent" (extractMessageText respMsg) llmLatency llmEndTime) + eStep <- (Right <$> runStep history) `catchError` (pure . Left) + llmEndTime <- liftIO getCurrentTime + let llmLatency = round (realToFrac (diffUTCTime llmEndTime llmStartTime) * 1000000 :: Double) :: Int + case eStep of + Left err -> do + dispatchCb agentCallbackManager (OnError "ReActAgent" (errorMessage err) llmEndTime) + throwError err + Right step -> case step of + AgentFinish finalMsg -> do + dispatchCb + agentCallbackManager + (OnLLMEnd "ReActAgent" (extractMessageText finalMsg) llmLatency llmEndTime) + let finalSteps = accSteps ++ [(step, [])] + pure $ AgentTrace finalSteps finalMsg (length finalSteps) + AgentAction respMsg tcs -> do + dispatchCb + agentCallbackManager + (OnLLMEnd "ReActAgent" (extractMessageText respMsg) llmLatency llmEndTime) + cancelledAfterStep <- checkCancelled + if cancelledAfterStep + then do + let finalSteps = accSteps ++ [(step, [])] + pure $ AgentTrace finalSteps respMsg (length finalSteps) + else case agentStopCondition of + Just cond + | cond respMsg -> do + let finalSteps = accSteps ++ [(step, [])] + pure $ AgentTrace finalSteps respMsg (length finalSteps) + _ -> do + let currentSig = toolSignature tcs + nextRepeatCount = + case mbLastSig of + Just prevSig | prevSig == currentSig -> repeatCount + 1 + _ -> 1 + case agentLoopDetection of + AbortOnRepeat maxR + | nextRepeatCount >= max 1 maxR -> do + let abortErr = + "Runaway loop detected: consecutive identical tool calls (" + <> T.pack (show nextRepeatCount) + <> " repetitions)" + now <- liftIO getCurrentTime + dispatchCb agentCallbackManager (OnError "ReActAgent" abortErr now) + throwError $ + agentError + abortErr + (Just "ReActAgent") + (Just "runLoopWith:loopDetection") + WarnOnRepeat maxR + | nextRepeatCount >= max 1 maxR -> do + let warnText tc = + "Loop warning: tool '" + <> toolCallName tc + <> "' called with identical arguments " + <> T.pack (show nextRepeatCount) + <> " times consecutively. Please change your approach, adjust arguments, or provide your final answer." + warnMsgs = + [ toolObservationMessage + (toolCallName tc) + (toolCallId tc) + (warnText tc) + | tc <- tcs + ] + newHistory = history ++ [respMsg] ++ warnMsgs + newSteps = accSteps ++ [(step, warnMsgs)] + go (Just currentSig) nextRepeatCount newHistory (maxIter - 1) newSteps + _ -> do + obsMsgs <- execToolsWithCancel tcs [] + let newHistory = history ++ [respMsg] ++ obsMsgs + newSteps = accSteps ++ [(step, obsMsgs)] + let _ = agent + go (Just currentSig) nextRepeatCount newHistory (maxIter - 1) newSteps - -- Check custom stop condition - case agentStopCondition of - Just cond | cond respMsg -> do - let allSteps = accSteps ++ [(step, [])] - pure $ AgentTrace allSteps respMsg - _ -> do - obsMsgs <- forM tcs executeTool - let newHistory = history ++ [respMsg] ++ obsMsgs - newSteps = accSteps ++ [(step, obsMsgs)] - go newHistory (maxIter - 1) newSteps +handleToolOutcome :: + (MonadIO n, MonadError LangchainError n) => + ReActAgent model m -> + ToolCall -> + (StreamEvent -> n ()) -> + Either LangchainError Text -> + Int -> + n Message +handleToolOutcome ReActAgent {..} tc emitStream eRes durationMicros = do + let tName = toolCallName tc + endTime <- liftIO getCurrentTime + case eRes of + Left err -> do + let errTxt = "Error executing tool " <> tName <> ": " <> errorMessage err + dispatchCb agentCallbackManager (OnError tName errTxt endTime) + emitStream $ ToolErrorEvent reactRunId tName err + when (agentToolErrorStrategy == FailOnError) $ + throwError $ + agentError errTxt (Just "ReActAgent") (Just $ "executeTool:" <> tName) + pure $ toolObservationMessage tName (toolCallId tc) errTxt + Right outTxt -> do + dispatchCb agentCallbackManager (OnToolEnd tName outTxt durationMicros endTime) + emitStream $ ToolEnd reactRunId tName (toJSON outTxt) + pure $ toolObservationMessage tName (toolCallId tc) outTxt - executeTool :: ToolCall -> m Message - executeTool tc = do - let tName = toolCallName tc - case agentCallbackManager of - Nothing -> pure () - Just mgr -> do - now <- liftIO getCurrentTime - dispatchEvent mgr (OnToolStart tName (toolCallArguments tc) now) +executeToolSync :: + forall model m. + (MonadIO m, MonadError LangchainError m) => + ReActAgent model m -> + ToolCall -> + m Message +executeToolSync agent@ReActAgent {..} tc = do + let tName = toolCallName tc + tools = allAgentTools agent + now <- liftIO getCurrentTime + dispatchCb agentCallbackManager (OnToolStart tName (toolCallArguments tc) now) + case find (\t -> toolName t == tName) tools of + Nothing -> do + let notFoundMsg = "Error: Tool not found: " <> tName + err = agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName) + handleToolOutcome agent tc (const (pure ())) (Left err) 0 + Just tool -> do + startTime <- liftIO getCurrentTime + eRes <- toolExecute tool (toolCallArguments tc) + endTime <- liftIO getCurrentTime + let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int + handleToolOutcome agent tc (const (pure ())) eRes durationMicros +executeToolStream :: + forall model m. + (MonadIO m, MonadError LangchainError m) => + ReActAgent model m -> + ToolCall -> + ConduitT () StreamEvent m Message +executeToolStream agent@ReActAgent {..} tc = do + let tName = toolCallName tc + now <- liftIO getCurrentTime + dispatchCb agentCallbackManager (OnToolStart tName (toolCallArguments tc) now) + yield $ ToolStart reactRunId tName (toolCallArguments tc) + case find (\st -> toolName (streamingToolDef st) == tName) agentStreamingTools of + Just sTool -> do + startTime <- liftIO getCurrentTime + eRes <- mapOutput (ToolChunk reactRunId tName) (streamingToolExecute sTool (toolCallArguments tc)) + endTime <- liftIO getCurrentTime + let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int + handleToolOutcome agent tc yield eRes durationMicros + Nothing -> case find (\t -> toolName t == tName) agentTools of Nothing -> do let notFoundMsg = "Error: Tool not found: " <> tName - case agentCallbackManager of - Nothing -> pure () - Just mgr -> do - now <- liftIO getCurrentTime - dispatchEvent mgr (OnError tName notFoundMsg now) - when (agentToolErrorStrategy == FailOnError) $ - throwError $ - agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName) - pure $ - (textMessage M.Tool notFoundMsg) - { M.messageName = Just tName - , M.messageToolId = Just (toolCallId tc) - } + err = agentError notFoundMsg (Just "ReActAgent") (Just $ "executeTool:" <> tName) + handleToolOutcome agent tc yield (Left err) 0 Just tool -> do startTime <- liftIO getCurrentTime - eRes <- toolExecute tool (toolCallArguments tc) + eRes <- lift $ toolExecute tool (toolCallArguments tc) endTime <- liftIO getCurrentTime let durationMicros = round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int - case eRes of - Left err -> do - let errTxt = "Error executing tool " <> tName <> ": " <> errorMessage err - case agentCallbackManager of - Nothing -> pure () - Just mgr -> dispatchEvent mgr (OnError tName errTxt endTime) - when (agentToolErrorStrategy == FailOnError) $ - throwError $ - agentError errTxt (Just "ReActAgent") (Just $ "executeTool:" <> tName) - pure $ - (textMessage M.Tool errTxt) - { M.messageName = Just tName - , M.messageToolId = Just (toolCallId tc) - } - Right outTxt -> do - case agentCallbackManager of - Nothing -> pure () - Just mgr -> dispatchEvent mgr (OnToolEnd tName outTxt durationMicros endTime) - pure $ - (textMessage M.Tool outTxt) - { M.messageName = Just tName - , M.messageToolId = Just (toolCallId tc) - } + handleToolOutcome agent tc yield eRes durationMicros + +{- | Execute the full ReAct reasoning loop, returning only the final answer. + +For the full reasoning trace, use 'runReActAgentWithTrace'. +For streaming execution, use 'streamReActAgent' or 'runReActAgentStream'. + +When the agent has a 'BaseMemory' attached: + +1. Existing conversation history is loaded from memory. +2. The user's input messages are appended. +3. After the loop finishes, both the user input and final answer are saved to memory. +-} +runReActAgent :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Configured ReAct agent instance + ReActAgent model m -> + -- | Input user messages for this turn + [Message] -> + -- | The final assistant answer message + m Message +runReActAgent agent userMsgs = traceFinalAnswer <$> runReActAgentWithTrace agent userMsgs + +{- | Execute the full ReAct reasoning loop, returning both the final answer +and the complete intermediate reasoning trace. + +Tool calls within a single step are executed in sequence within the @m@ monad, +and the observations are fed back to the LLM for the next reasoning step. + +The trace includes every 'AgentStep' and the observation messages produced +at each iteration, which is invaluable for debugging and observability. +-} +runReActAgentWithTrace :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Configured ReAct agent instance + ReActAgent model m -> + -- | Input user messages for this turn + [Message] -> + -- | Full reasoning trace containing all intermediate steps and final answer + m AgentTrace +runReActAgentWithTrace agent@ReActAgent {..} userMsgs = do + initialHistory <- buildInitialHistory agent userMsgs "runReActAgentWithTrace" + let tools = allAgentTools agent + trace <- + runLoopWith + Nothing + agent + (\hist -> reactStep agentModel tools hist agentModelConfig) + (executeToolSync agent) + initialHistory + agentMaxIterations + [] + persistTurnMemory agent userMsgs (traceFinalAnswer trace) + pure trace + +streamReActAgentTraceWith :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + Maybe CancelToken -> + ReActAgent model m -> + [Message] -> + ConduitT () StreamEvent m AgentTrace +streamReActAgentTraceWith mbTok agent@ReActAgent {..} userMsgs = do + initialHistory <- lift $ buildInitialHistory agent userMsgs "streamReActAgentWithTrace" + let tools = allAgentTools agent + yield $ ChainStart reactRunId "ReActAgent" (toJSON initialHistory) + trace <- + runLoopWith + mbTok + agent + (\hist -> reactStepStreamWith mbTok agentModel tools hist agentModelConfig) + (executeToolStream agent) + initialHistory + agentMaxIterations + [] + cancelled <- maybe (pure False) isCancelled mbTok + unless cancelled $ do + lift $ persistTurnMemory agent userMsgs (traceFinalAnswer trace) + yield $ ChainEnd reactRunId "ReActAgent" (toJSON (traceFinalAnswer trace)) + pure trace + +{- | Execute the full ReAct reasoning loop as a Conduit stream of 'StreamEvent's, +returning the complete 'AgentTrace'. + +Emits 'ChainStart', followed by 'LLMStart' / 'LLMChunk' / 'LLMEnd' for each +reasoning turn and 'ToolStart' / 'ToolChunk' / 'ToolEnd' / 'ToolErrorEvent' for +each tool invocation, concluding with 'ChainEnd'. +-} +streamReActAgentWithTrace :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Configured ReAct agent instance + ReActAgent model m -> + -- | Input user messages for this turn + [Message] -> + -- | Event stream returning the full reasoning trace + ConduitT () StreamEvent m AgentTrace +streamReActAgentWithTrace = streamReActAgentTraceWith Nothing + +{- | Execute the full ReAct reasoning loop as an 'EventStream'. + +Suitable for composition with 'collectEvents', 'runStreamWith', or any Conduit sink. +-} +streamReActAgent :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Configured ReAct agent instance + ReActAgent model m -> + -- | Input user messages for this turn + [Message] -> + -- | Stream of agent lifecycle, LLM, and tool events + EventStream m +streamReActAgent agent userMsgs = do + _ <- streamReActAgentWithTrace agent userMsgs + pure () + +-- | Outcome of an interruptible streaming ReAct turn. +data InterruptibleResult + = -- | Turn completed normally with the final assistant message. + Completed !Message + | -- | Turn was cancelled via 'CancelToken', returning any partial assistant text emitted. + Cancelled !Message + deriving (Eq, Show) + +{- | Execute an interruptible streaming ReAct turn using a cooperative 'CancelToken'. + +Safe for @LangchainT () IO@ callers without @ResourceT@. When cancelled mid-turn, +upstream streaming stops promptly, conversation memory is left unmodified, and +'Cancelled' returns the partial assistant message streamed so far. +-} +runReActAgentStreamInterruptible :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + CancelToken -> + ReActAgent model m -> + [Message] -> + (StreamEvent -> m ()) -> + m InterruptibleResult +runReActAgentStreamInterruptible tok agent userMsgs onEvent = do + cancelledStart <- isCancelled tok + if cancelledStart + then pure $ Cancelled (assistantMessage "") + else do + chunksRef <- liftIO $ newIORef ([] :: [Text]) + let trackAndForward ev = do + liftIO $ case ev of + LLMStart {} -> writeIORef chunksRef [] + LLMChunk _ txt _ | not (T.null txt) -> modifyIORef' chunksRef (txt :) + LLMEnd _ msg _ -> writeIORef chunksRef [extractMessageText msg] + _ -> pure () + onEvent ev + trace <- + runConduit + ( streamReActAgentTraceWith (Just tok) agent userMsgs + `fuseUpstream` CL.mapM_ trackAndForward + ) + cancelledEnd <- isCancelled tok + if cancelledEnd + then do + partialTxt <- liftIO (T.concat . reverse <$> readIORef chunksRef) + pure $ Cancelled (assistantMessage partialTxt) + else pure $ Completed (traceFinalAnswer trace) + +{- | Execute the ReAct agent in streaming mode, invoking a callback for every +'StreamEvent' as it arrives and returning the final assistant 'Message'. +-} +runReActAgentStream :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Configured ReAct agent instance + ReActAgent model m -> + -- | Input user messages for this turn + [Message] -> + -- | Callback invoked on each 'StreamEvent' in real time + (StreamEvent -> m ()) -> + -- | The final assistant answer message + m Message +runReActAgentStream agent userMsgs onEvent = + traceFinalAnswer + <$> runConduit (streamReActAgentWithTrace agent userMsgs `fuseUpstream` CL.mapM_ onEvent) + +-- | JSON Schema describing the input expected by agent handoff tools. +handoffSchema :: Value +handoffSchema = + object + [ "type" .= ("object" :: Text) + , "properties" + .= object + [ "task" + .= object + [ "type" .= ("string" :: Text) + , "description" .= ("The task or query for the specialist agent" :: Text) + ] + ] + , "required" .= (["task"] :: [Text]) + ] + +-- | Extract the task query string from tool arguments. +extractAgentTask :: Text -> Value -> Either LangchainError Text +extractAgentTask toolNm = \case + String s | not (T.null (T.strip s)) -> Right s + Object o -> case parseEither (.:? "task") o of + Right (Just task) | not (T.null (T.strip task)) -> Right task + _ -> case parseEither (.:? "query") o of + Right (Just q) | not (T.null (T.strip q)) -> Right q + _ -> case parseEither (.:? "prompt") o of + Right (Just p) | not (T.null (T.strip p)) -> Right p + _ -> + Left $ + toolError + ("Missing or invalid 'task' parameter for agent tool: " <> toolNm) + (Just toolNm) + Nothing + _ -> + Left $ + toolError + ("Invalid arguments object for agent tool: " <> toolNm) + (Just toolNm) + Nothing + +{- | Wrap a 'ReActAgent' as a callable 'Tool' that delegates tasks to the sub-agent. + +Accepts @{ "task": "" }@ (or a raw JSON string) and returns the sub-agent's +final answer text. Bounded by 'agentMaxDelegationDepth' to prevent unbounded recursion. +-} +asAgentTool :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Tool name (e.g., "researcher") + Text -> + -- | Tool description explaining the specialist agent's role + Text -> + -- | Specialist sub-agent to delegate tasks to + ReActAgent model m -> + -- | Wrapped tool ready to be attached to a supervisor agent + Tool m +asAgentTool name desc subAgent = + createTool name desc handoffSchema $ \args -> do + case extractAgentTask name args of + Left err -> pure (Left err) + Right task + | agentMaxDelegationDepth subAgent <= 0 -> + pure $ + Left $ + agentError + ("Maximum agent delegation depth exceeded for sub-agent '" <> name <> "'") + (Just "ReActAgent") + (Just "asAgentTool") + | otherwise -> do + let delegatedAgent = + subAgent {agentMaxDelegationDepth = agentMaxDelegationDepth subAgent - 1} + eRes <- (Right <$> runReActAgent delegatedAgent [userMessage task]) `catchError` (pure . Left) + pure $ fmap extractMessageText eRes + +{- | Wrap a 'ReActAgent' as a 'StreamingTool' forwarding intermediate progress chunks to the caller. + +Emits incremental sub-agent text and tool chunks downstream wrapped as 'ToolChunk' events. +-} +asStreamingAgentTool :: + forall model m. + (ToolBinder model m, MonadIO m, MonadError LangchainError m) => + -- | Tool name (e.g., "researcher") + Text -> + -- | Tool description explaining the specialist agent's role + Text -> + -- | Specialist sub-agent to delegate tasks to + ReActAgent model m -> + -- | Wrapped streaming tool emitting incremental sub-agent chunks + StreamingTool m +asStreamingAgentTool name desc subAgent = + createStreamingTool name desc handoffSchema $ \args -> do + case extractAgentTask name args of + Left err -> pure (Left err) + Right task + | agentMaxDelegationDepth subAgent <= 0 -> + pure $ + Left $ + agentError + ("Maximum agent delegation depth exceeded for sub-agent '" <> name <> "'") + (Just "ReActAgent") + (Just "asStreamingAgentTool") + | otherwise -> do + let delegatedAgent = + subAgent {agentMaxDelegationDepth = agentMaxDelegationDepth subAgent - 1} + ansRef <- liftIO $ newIORef "" + let captureSub = do + trace <- streamReActAgentWithTrace delegatedAgent [userMessage task] + liftIO $ writeIORef ansRef (extractMessageText (traceFinalAnswer trace)) + forwardEvents = do + mEv <- await + case mEv of + Nothing -> pure () + Just ev -> do + case ev of + LLMChunk _ txt Nothing | not (T.null txt) -> yield txt + ToolChunk _ _ chunk | not (T.null chunk) -> yield chunk + _ -> pure () + forwardEvents + catchError + ( do + captureSub .| forwardEvents + ans <- liftIO $ readIORef ansRef + pure $ Right ans + ) + (pure . Left) diff --git a/src/Langchain/Eval.hs b/src/Langchain/Eval.hs new file mode 100644 --- /dev/null +++ b/src/Langchain/Eval.hs @@ -0,0 +1,308 @@ +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} + +{- | +Module : Langchain.Eval +Description : Lightweight evaluation harness for chains, agents, and models +Copyright : (c) 2025-2026 Tushar Adhatrao +License : MIT +Maintainer : Tushar Adhatrao +Stability : experimental + +Quantitative evaluation and regression testing framework for LLM applications. +Provides typed datasets ('EvalCase'), deterministic scorers (exact match, substring, +tool call trace accuracy), LLM-as-a-judge scoring, and JSON evaluation report generation. +-} +module Langchain.Eval + ( -- * Core Types + EvalCase (..) + , EvalScore (..) + , Scorer (..) + , createScorer + , CaseResult (..) + , CaseScoreResult (..) + , EvalReport (..) + + -- * Deterministic Scorers + , exactMatchScorer + , containsScorer + , toolCallExactScorer + + -- * Model-based Scorers + , llmJudgeScorer + + -- * Evaluation Runner + , runEvalSuite + ) where + +import Control.Monad.Except (MonadError, catchError) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Aeson + ( FromJSON (..) + , ToJSON (..) + , Value (..) + , decode + , withObject + , (.:) + , (.:?) + ) +import Data.Aeson.Types (parseEither) +import qualified Data.ByteString.Lazy as LBS +import Data.Map.Strict (Map) +import Data.Maybe (fromMaybe) +import Data.Text (Text) +import qualified Data.Text as T +import qualified Data.Text.Encoding as TE +import Data.Time.Clock (diffUTCTime, getCurrentTime) +import GHC.Generics (Generic) + +import Langchain.Agent.ReAct (AgentStep (..), AgentTrace (..)) +import Langchain.Error (LangchainError, errorMessage) +import Langchain.Model (ChatModel (..)) +import Langchain.Model.Types (extractMessageText, toolCallName, userMessage) + +-- | A single evaluation test case with input, optional reference target, and metadata. +data EvalCase i o = EvalCase + { evalInput :: !i + -- ^ Input supplied to the subject under test + , evalExpected :: !(Maybe o) + -- ^ Optional ground-truth or expected reference output + , evalMetadata :: !(Maybe (Map Text Text)) + -- ^ Optional contextual tags or metadata + } + deriving (Show, Eq, Generic, ToJSON, FromJSON) + +-- | Assessment produced by a 'Scorer' on an individual test case. +data EvalScore = EvalScore + { scoreValue :: !Double + -- ^ Normalized score between 0.0 and 1.0 + , scorePassed :: !Bool + -- ^ Whether the case satisfied the threshold or assertion + , scoreReason :: !(Maybe Text) + -- ^ Diagnostic or qualitative reasoning for the score + } + deriving (Show, Eq, Generic, ToJSON, FromJSON) + +-- | Named evaluation metric evaluating actual output against a test case. +data Scorer i o m = Scorer + { scorerName :: !Text + -- ^ Unique identifier for this evaluation metric + , scoreCase :: EvalCase i o -> o -> m EvalScore + -- ^ Evaluation function producing 'EvalScore' + } + +-- | Construct a custom 'Scorer'. +createScorer :: Text -> (EvalCase i o -> o -> m EvalScore) -> Scorer i o m +createScorer = Scorer + +-- | Score result produced by a specific named scorer. +data CaseScoreResult = CaseScoreResult + { caseScoreName :: !Text + , caseScore :: !EvalScore + } + deriving (Show, Eq, Generic, ToJSON, FromJSON) + +-- | Outcome of running all scorers on a single test case. +data CaseResult = CaseResult + { caseInput :: !Value + , caseExpected :: !(Maybe Value) + , caseActual :: !(Either Text Value) + , caseScores :: ![CaseScoreResult] + , caseLatencyMicros :: !Int + , casePassed :: !Bool + } + deriving (Show, Eq, Generic, ToJSON, FromJSON) + +-- | Aggregated summary report of an evaluation suite run. +data EvalReport = EvalReport + { reportTotalCases :: !Int + -- ^ Total number of evaluated cases + , reportPassedCases :: !Int + -- ^ Number of cases that passed all scorers + , reportPassRate :: !Double + -- ^ Ratio of passed cases (0.0 to 1.0) + , reportMeanScore :: !Double + -- ^ Average score value across all evaluations (0.0 to 1.0) + , reportMeanLatencyMicros :: !Int + -- ^ Average execution latency per case in microseconds + , reportCases :: ![CaseResult] + -- ^ Detailed per-case scoring breakdown + } + deriving (Show, Eq, Generic, ToJSON, FromJSON) + +-- | Deterministic scorer verifying exact equality between actual and expected output. +exactMatchScorer :: (Eq o, Show o, Monad m) => Scorer i o m +exactMatchScorer = createScorer "exact_match" $ \c actual -> + case evalExpected c of + Nothing -> pure $ EvalScore 1.0 True (Just "No expected output specified") + Just expected -> + if actual == expected + then pure $ EvalScore 1.0 True (Just "Exact match") + else + pure $ + EvalScore + 0.0 + False + (Just $ "Expected: " <> T.pack (show expected) <> ", Actual: " <> T.pack (show actual)) + +-- | Deterministic scorer checking if expected text is contained within actual text. +containsScorer :: Monad m => Scorer Text Text m +containsScorer = createScorer "contains" $ \c actual -> + case evalExpected c of + Nothing -> pure $ EvalScore 1.0 True (Just "No expected substring specified") + Just expectedSub -> + if expectedSub `T.isInfixOf` actual + then pure $ EvalScore 1.0 True (Just "Substring found") + else + pure $ + EvalScore + 0.0 + False + (Just $ "Substring '" <> expectedSub <> "' not found in actual output") + +-- | Deterministic scorer verifying that an 'AgentTrace' invoked an exact sequence of tools. +toolCallExactScorer :: Monad m => [Text] -> Scorer i AgentTrace m +toolCallExactScorer expectedTools = createScorer "tool_call_exact" $ \_ trace -> do + let actualTools = + [ toolCallName tc + | (AgentAction _ tcs, _) <- traceSteps trace + , tc <- tcs + ] + if actualTools == expectedTools + then pure $ EvalScore 1.0 True (Just "Tool call sequence matched exactly") + else + pure $ + EvalScore + 0.0 + False + ( Just $ + "Expected tool sequence: " + <> T.pack (show expectedTools) + <> ", but got: " + <> T.pack (show actualTools) + ) + +-- | LLM-as-a-judge scorer evaluating output against a custom qualitative rubric. +llmJudgeScorer :: + forall model m. + (ChatModel model, MonadIO m, MonadError LangchainError m) => + model -> + Text -> + Scorer Text Text m +llmJudgeScorer model rubric = createScorer "llm_judge" $ \c actual -> do + let inputTxt = evalInput c + expectedTxt = fromMaybe "" (evalExpected c) + prompt = + "You are an expert AI evaluator. Assess the following actual output against the rubric.\n\n" + <> "Evaluation Rubric: " + <> rubric + <> "\n\n" + <> "Test Input: " + <> inputTxt + <> "\n" + <> "Reference Target: " + <> expectedTxt + <> "\n" + <> "Actual Output: " + <> actual + <> "\n\n" + <> "Reply with a JSON object: {\"passed\": true/false, \"score\": 0.0-1.0, \"reason\": \"\"}" + eResp <- (Right <$> invoke model [userMessage prompt] Nothing) `catchError` (pure . Left) + case eResp of + Left err -> pure $ EvalScore 0.0 False (Just $ "LLM judge invocation failed: " <> errorMessage err) + Right resp -> do + let outTxt = extractMessageText resp + case parseJudgeJson outTxt of + Just (passed, sc, rsn) -> pure $ EvalScore sc passed (Just rsn) + Nothing -> + let passed = not ("fail" `T.isInfixOf` T.toLower outTxt) + sc = if passed then 1.0 else 0.0 + in pure $ EvalScore sc passed (Just outTxt) + +parseJudgeJson :: Text -> Maybe (Bool, Double, Text) +parseJudgeJson txt = do + val <- decode (LBS.fromStrict (TE.encodeUtf8 txt)) + case parseEither parser val of + Right res -> Just res + Left _ -> Nothing + where + parser = withObject "JudgeResponse" $ \o -> do + passed <- o .: "passed" + sc <- + o .:? "score" >>= \case + Just s -> pure s + Nothing -> pure (if passed then 1.0 else 0.0) + rsn <- + o .:? "reason" >>= \case + Just r -> pure r + Nothing -> pure "" + pure (passed, sc, rsn) + +-- | Execute an evaluation suite over a set of test cases and scorers. +runEvalSuite :: + forall i o m. + (MonadIO m, ToJSON i, ToJSON o) => + (i -> m (Either LangchainError o)) -> + [Scorer i o m] -> + [EvalCase i o] -> + m EvalReport +runEvalSuite runner scorers cases = do + caseResults <- mapM evaluateCase cases + let total = length caseResults + passedCount = length (filter casePassed caseResults) + passRate = if total == 0 then 1.0 else fromIntegral passedCount / fromIntegral total + allScores = [scoreValue (caseScore sr) | cr <- caseResults, sr <- caseScores cr] + meanScore = + if null allScores + then if total == 0 then 1.0 else 0.0 + else sum allScores / fromIntegral (length allScores) + allLatencies = map caseLatencyMicros caseResults + meanLatency = + if null allLatencies + then 0 + else sum allLatencies `div` total + pure + EvalReport + { reportTotalCases = total + , reportPassedCases = passedCount + , reportPassRate = passRate + , reportMeanScore = meanScore + , reportMeanLatencyMicros = meanLatency + , reportCases = caseResults + } + where + evaluateCase c = do + startTime <- liftIO getCurrentTime + eActual <- runner (evalInput c) + endTime <- liftIO getCurrentTime + let latencyMicros = + round (realToFrac (diffUTCTime endTime startTime) * 1000000 :: Double) :: Int + case eActual of + Left err -> do + let failedScore = EvalScore 0.0 False (Just $ errorMessage err) + scores = [CaseScoreResult (scorerName s) failedScore | s <- scorers] + pure + CaseResult + { caseInput = toJSON (evalInput c) + , caseExpected = fmap toJSON (evalExpected c) + , caseActual = Left (errorMessage err) + , caseScores = scores + , caseLatencyMicros = latencyMicros + , casePassed = False + } + Right actual -> do + scores <- mapM (\s -> CaseScoreResult (scorerName s) <$> scoreCase s c actual) scorers + let caseOk = all (scorePassed . caseScore) scores + pure + CaseResult + { caseInput = toJSON (evalInput c) + , caseExpected = fmap toJSON (evalExpected c) + , caseActual = Right (toJSON actual) + , caseScores = scores + , caseLatencyMicros = latencyMicros + , casePassed = caseOk + } diff --git a/src/Langchain/Prelude.hs b/src/Langchain/Prelude.hs --- a/src/Langchain/Prelude.hs +++ b/src/Langchain/Prelude.hs @@ -56,6 +56,7 @@ , LLMChunk , LLMEnd , ToolStart + , ToolChunk , ToolEnd , ToolErrorEvent , ChainStart @@ -65,6 +66,18 @@ ) , TokenUsage (..) , EventStream + , StreamM + , ChatStream + , CancelToken + , newCancelToken + , triggerCancel + , isCancelled + , liftChatStream + , liftChatStreamResource + , liftChatStreamWithCancel + , tapStreamMessage + , consumeStreamWith + , runStreamWith , collectEvents , printEvents @@ -100,6 +113,11 @@ , Tool (..) , createTool , toolToValue + , StreamingTool (..) + , createStreamingTool + , fromTool + , streamToolCall + , streamStreamingToolCall , ToolBinder (..) , DeriveToolSchema (..) , deriveToolParametersSchema @@ -108,10 +126,14 @@ , PlanStep (..) , Plan (..) , StepExecutor (..) + , StreamingStepExecutor (..) , PlanAndExecuteAgent (..) , newPlanAndExecuteAgent , newPlanAndExecuteAgentWithTools , runPlanAndExecute + , streamPlanAndExecute + , streamPlanAndExecuteWithResult + , runPlanAndExecuteStream -- * Guardrails & Safety , GuardrailResult (..) @@ -293,6 +315,7 @@ , AgentStep (..) , AgentTrace (..) , ToolErrorStrategy (..) + , LoopDetectionStrategy (..) , defaultReActAgent , withMemory , withoutMemory @@ -302,9 +325,21 @@ , withCallbackManager , withStopCondition , withToolErrorStrategy + , withStreamingTools + , withLoopDetection + , withMaxConsecutiveIdenticalToolCalls + , withMaxDelegationDepth + , asAgentTool + , asStreamingAgentTool , reactStep , runReActAgent , runReActAgentWithTrace + , reactStepStream + , streamReActAgent + , streamReActAgentWithTrace + , runReActAgentStream + , InterruptibleResult (..) + , runReActAgentStreamInterruptible -- * Tool Suite & Security , ToolTier (..) @@ -313,6 +348,7 @@ , readOnlySecurityPolicy , workspaceSandboxPolicy , withSecurity + , withStreamingSecurity , newFileSystemSuite , findFilesTool , grepSearchTool @@ -325,6 +361,8 @@ , commandStatusTool , commandKillTool , shellTool + , streamingShellTool + , streamingShellToolWithCancel -- * Hybrid Retrieval & BM25 , BM25Index (..) @@ -339,6 +377,20 @@ , searchHybrid , searchHybridWithScores , reciprocalRankFusion + + -- * Evaluation & Regression Testing + , EvalCase (..) + , EvalScore (..) + , Scorer (..) + , createScorer + , CaseResult (..) + , CaseScoreResult (..) + , EvalReport (..) + , exactMatchScorer + , containsScorer + , toolCallExactScorer + , llmJudgeScorer + , runEvalSuite ) where import Langchain.Agent.PlanAndExecute @@ -351,6 +403,7 @@ import Langchain.DocumentLoader.FileLoader import Langchain.Embeddings.Core import Langchain.Error +import Langchain.Eval import Langchain.Guardrail.Core import Langchain.Memory import Langchain.Model @@ -377,6 +430,6 @@ import Langchain.Tool.GenericSchema import Langchain.Tool.Process import Langchain.Tool.Security -import Langchain.Tool.Shell (shellTool) +import Langchain.Tool.Shell (shellTool, streamingShellTool, streamingShellToolWithCancel) import Langchain.VectorStore.Core import Langchain.VectorStore.InMemory diff --git a/src/Langchain/Stream.hs b/src/Langchain/Stream.hs --- a/src/Langchain/Stream.hs +++ b/src/Langchain/Stream.hs @@ -21,32 +21,60 @@ , StreamCallback , StreamSource , callbackSource + , liftChatStream + , liftChatStreamResource + , CancelToken + , newCancelToken + , triggerCancel + , isCancelled + , waitCancel + , liftChatStreamWithCancel + , tapStreamMessage + , consumeStreamWith + , runStreamWith , collectEvents , printEvents ) where -import Control.Concurrent.Async (async, cancel) +import Control.Concurrent.Async (Async, AsyncCancelled (..), async, cancel, wait) import Control.Concurrent.STM - ( atomically + ( STM + , TBQueue + , TMVar + , atomically , newEmptyTMVarIO , newTBQueueIO , orElse , putTMVar , readTBQueue , readTMVar + , tryPutTMVar + , tryReadTMVar , writeTBQueue ) -import Control.Exception (finally) -import Control.Monad.Except (ExceptT, runExceptT) -import Control.Monad.IO.Class (liftIO) -import Control.Monad.Trans.Resource (ResourceT, runResourceT) +import Control.Exception + ( Exception (displayException) + , SomeException + , catch + , finally + , fromException + , onException + , throwIO + ) +import Control.Monad (void) +import Control.Monad.Except (ExceptT, MonadError (throwError), runExceptT) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Trans.Resource (MonadResource, ResourceT, runResourceT) import Data.Aeson (FromJSON, ToJSON, Value) -import Data.Conduit (ConduitT, bracketP, runConduit, yield, (.|)) +import Data.Conduit (ConduitT, Void, await, bracketP, fuseUpstream, runConduit, yield, (.|)) import qualified Data.Conduit.List as CL +import Data.IORef (newIORef, readIORef, writeIORef) +import Data.Maybe (isJust) import Data.Text (Text) +import qualified Data.Text as T import GHC.Generics (Generic) -import Langchain.Error (LangchainError) -import Langchain.Model.Types (Message, ToolCall) +import Langchain.Error (LangchainError, llmError) +import Langchain.Model.Types (Message (..), ToolCall, assistantMessage) -- | Token usage accounting for LLM execution. data TokenUsage = TokenUsage @@ -84,6 +112,12 @@ , toolName :: Text , toolInput :: Value } + | -- | Tool incremental streaming output chunk + ToolChunk + { runId :: Text + , toolName :: Text + , toolChunkText :: Text + } | -- | Tool execution completion ToolEnd { runId :: Text @@ -159,6 +193,176 @@ Nothing -> pure () third (_, _, worker) = worker + +-- | Cooperative single-shot cancellation token backed by an STM 'TMVar'. +newtype CancelToken = CancelToken (TMVar ()) + deriving (Eq) + +-- | Allocate a new untriggered 'CancelToken'. +newCancelToken :: MonadIO m => m CancelToken +newCancelToken = CancelToken <$> liftIO newEmptyTMVarIO + +-- | Signal cancellation on a 'CancelToken' (idempotent). +triggerCancel :: MonadIO m => CancelToken -> m () +triggerCancel (CancelToken var) = liftIO . atomically . void $ tryPutTMVar var () + +-- | Check whether a 'CancelToken' has been triggered. +isCancelled :: MonadIO m => CancelToken -> m Bool +isCancelled (CancelToken var) = liftIO . atomically $ isJust <$> tryReadTMVar var + +waitCancel :: CancelToken -> STM () +waitCancel (CancelToken var) = readTMVar var + +startChatWorker :: + ChatStream -> + IO (TBQueue StreamEvent, TMVar (Maybe LangchainError), Async ()) +startChatWorker chatStream = do + q <- newTBQueueIO 64 + done <- newEmptyTMVarIO + errRef <- newIORef Nothing + w <- + async $ + ( do + eRes <- + runResourceT $ + runExceptT $ + runConduit $ + chatStream .| CL.mapM_ (liftIO . atomically . writeTBQueue q) + case eRes of + Right () -> pure () + Left err -> writeIORef errRef (Just err) + ) + `catch` ( \ex -> case fromException ex of + Just AsyncCancelled -> throwIO (ex :: SomeException) + Nothing -> + writeIORef + errRef + (Just $ llmError (T.pack (displayException ex)) (Just "Stream") Nothing) + ) + `finally` (readIORef errRef >>= atomically . void . tryPutTMVar done) + pure (q, done, w) + +consumeChatWorker :: + (MonadIO m, MonadError LangchainError m) => + Maybe CancelToken -> + (TBQueue StreamEvent, TMVar (Maybe LangchainError), Async ()) -> + EventStream m +consumeChatWorker mbTok (queue, doneVar, worker) = loop + where + loop = do + next <- + liftIO $ + atomically + ( case mbTok of + Just tok -> + (Left Nothing <$ waitCancel tok) + `orElse` (Right <$> readTBQueue queue) + `orElse` (Left . Just <$> readTMVar doneVar) + Nothing -> + (Right <$> readTBQueue queue) + `orElse` (Left . Just <$> readTMVar doneVar) + ) + `onException` cancel worker + case next of + Right ev -> do + yield ev + loop + Left Nothing -> + liftIO $ cancel worker + Left (Just Nothing) -> + liftIO $ wait worker + Left (Just (Just err)) -> do + liftIO $ cancel worker + throwError err + +{- | Lift a resource-safe 'ChatStream' into any 'MonadIO' and 'MonadError LangchainError' +'EventStream'. Events are forwarded incrementally through a bounded STM queue, +and cancelling or throwing an exception cancels the upstream stream worker promptly. +-} +liftChatStream :: + (MonadIO m, MonadError LangchainError m) => + ChatStream -> + EventStream m +liftChatStream chatStream = do + st <- liftIO $ startChatWorker chatStream + consumeChatWorker Nothing st + +{- | Lift a resource-safe 'ChatStream' into any 'MonadResource' and +'MonadError LangchainError' 'EventStream'. Uses 'bracketP' so stopping +consumption early (e.g. @src .| CL.take 1@) cancels the upstream producer +immediately. +-} +liftChatStreamResource :: + (MonadResource m, MonadError LangchainError m) => + ChatStream -> + EventStream m +liftChatStreamResource chatStream = + bracketP (startChatWorker chatStream) (\(_, _, w) -> cancel w) (consumeChatWorker Nothing) + +{- | Lift a resource-safe 'ChatStream' with cooperative cancellation via +'CancelToken'. When cancelled, the upstream stream worker is terminated promptly +and the Conduit exits cleanly so downstream accumulators preserve partial text. +-} +liftChatStreamWithCancel :: + (MonadIO m, MonadError LangchainError m) => + CancelToken -> + ChatStream -> + EventStream m +liftChatStreamWithCancel tok chatStream = do + st <- liftIO $ startChatWorker chatStream + consumeChatWorker (Just tok) st + +{- | Pass every 'StreamEvent' downstream unchanged while accumulating the final 'Message' +from the last 'LLMEnd' event (or synthesizing one from accumulated 'LLMChunk's if +no 'LLMEnd' event was emitted). +-} +tapStreamMessage :: Monad m => ConduitT StreamEvent StreamEvent m Message +tapStreamMessage = loop [] [] Nothing + where + loop accText accTools mbFinal = do + mEv <- await + case mEv of + Nothing -> + pure $ case mbFinal of + Just msg -> + case (messageToolCalls msg, reverse accTools) of + (Nothing, ts@(_ : _)) -> msg {messageToolCalls = Just ts} + _ -> msg + Nothing -> + let fullText = T.concat (reverse accText) + ts = reverse accTools + in (assistantMessage fullText) + { messageToolCalls = if null ts then Nothing else Just ts + } + Just ev -> do + yield ev + case ev of + LLMStart {} -> loop [] [] Nothing + LLMChunk {..} -> + let nextText = if T.null chunkText then accText else chunkText : accText + nextTools = maybe accTools (: accTools) toolCallDelta + in loop nextText nextTools mbFinal + LLMEnd {..} -> + loop accText accTools (Just finalMessage) + _ -> loop accText accTools mbFinal + +{- | Consume a 'StreamEvent' pipeline, invoking a handler for every event, and +return the final 'Message' from the last 'LLMEnd' event (or synthesize one from +accumulated 'LLMChunk's if no 'LLMEnd' event was emitted). +-} +consumeStreamWith :: + Monad m => + (StreamEvent -> m ()) -> + ConduitT StreamEvent Void m Message +consumeStreamWith onEvent = tapStreamMessage `fuseUpstream` CL.mapM_ onEvent + +-- | Run an 'EventStream' with an event handler and return the resulting 'Message'. +runStreamWith :: + Monad m => + (StreamEvent -> m ()) -> + EventStream m -> + m Message +runStreamWith onEvent src = runConduit (src .| consumeStreamWith onEvent) -- | Collect all emitted events from a stream into a list. collectEvents :: Monad m => EventStream m -> m [StreamEvent] diff --git a/src/Langchain/Tool.hs b/src/Langchain/Tool.hs --- a/src/Langchain/Tool.hs +++ b/src/Langchain/Tool.hs @@ -14,12 +14,20 @@ ( Tool (..) , createTool , toolToValue + , StreamingTool (..) + , createStreamingTool + , fromTool + , streamToolCall + , streamStreamingToolCall ) where +import Control.Monad.Trans (lift) import Data.Aeson +import Data.Conduit (ConduitT, await, fuseUpstream, mapOutput, runConduit, yield) import Data.Text (Text) import Langchain.Error (LangchainError) +import Langchain.Stream (StreamEvent (ToolChunk, ToolEnd, ToolErrorEvent, ToolStart)) -- | Effect-polymorphic Tool abstraction data Tool m = Tool @@ -63,3 +71,90 @@ , "parameters" .= toolSchema ] ] + +-- | Effect-polymorphic streaming Tool abstraction supporting incremental 'Text' output chunks. +data StreamingTool m = StreamingTool + { streamingToolDef :: Tool m + -- ^ Underlying tool specification and synchronous fallback executor + , streamingToolExecute :: Value -> ConduitT () Text m (Either LangchainError Text) + -- ^ Monadic streaming execution yielding output chunks and returning final text output + } + +instance Show (StreamingTool m) where + show st = "StreamingTool { toolName = " ++ show (toolName (streamingToolDef st)) ++ " }" + +-- | Construct a 'StreamingTool' from a name, description, schema, and streaming executor. +createStreamingTool :: + Monad m => + Text -> + Text -> + Value -> + (Value -> ConduitT () Text m (Either LangchainError Text)) -> + StreamingTool m +createStreamingTool name desc schema streamExec = + StreamingTool + { streamingToolDef = + createTool name desc schema $ \args -> + runConduit (streamExec args `fuseUpstream` drainSink) + , streamingToolExecute = streamExec + } + where + drainSink = do + mItem <- await + case mItem of + Nothing -> pure () + Just _ -> drainSink + +-- | Lift a standard 'Tool' into a 'StreamingTool' that emits its full output as a single chunk. +fromTool :: Monad m => Tool m -> StreamingTool m +fromTool t = + StreamingTool + { streamingToolDef = t + , streamingToolExecute = \args -> do + res <- lift $ toolExecute t args + case res of + Right out -> yield out >> pure (Right out) + Left err -> pure (Left err) + } + +{- | Execute a 'Tool' within a 'StreamEvent' pipeline, emitting 'ToolStart' +followed by 'ToolEnd' or 'ToolErrorEvent'. +-} +streamToolCall :: + Monad m => + Text -> + Tool m -> + Value -> + ConduitT () StreamEvent m (Either LangchainError Text) +streamToolCall rId tool args = do + let tName = toolName tool + yield $ ToolStart rId tName args + eRes <- lift $ toolExecute tool args + case eRes of + Left err -> do + yield $ ToolErrorEvent rId tName err + pure (Left err) + Right out -> do + yield $ ToolEnd rId tName (toJSON out) + pure (Right out) + +{- | Execute a 'StreamingTool' within a 'StreamEvent' pipeline, emitting +'ToolStart', 'ToolChunk' for each streamed text chunk, and 'ToolEnd' or 'ToolErrorEvent'. +-} +streamStreamingToolCall :: + Monad m => + Text -> + StreamingTool m -> + Value -> + ConduitT () StreamEvent m (Either LangchainError Text) +streamStreamingToolCall rId sTool args = do + let tName = toolName (streamingToolDef sTool) + yield $ ToolStart rId tName args + eRes <- mapOutput (ToolChunk rId tName) (streamingToolExecute sTool args) + case eRes of + Left err -> do + yield $ ToolErrorEvent rId tName err + pure (Left err) + Right out -> do + yield $ ToolEnd rId tName (toJSON out) + pure (Right out) diff --git a/src/Langchain/Tool/Security.hs b/src/Langchain/Tool/Security.hs --- a/src/Langchain/Tool/Security.hs +++ b/src/Langchain/Tool/Security.hs @@ -38,6 +38,7 @@ -- * Tool Security Combinator , withSecurity + , withStreamingSecurity ) where import Control.Monad.IO.Class (MonadIO, liftIO) @@ -51,8 +52,9 @@ import System.Directory (canonicalizePath, doesDirectoryExist, doesFileExist) import System.FilePath (isAbsolute, normalise, pathSeparator, takeDirectory, takeFileName, ()) +import Control.Monad.Trans (lift) import Langchain.Error (LangchainError, toolError) -import Langchain.Tool (Tool (..)) +import Langchain.Tool (StreamingTool (..), Tool (..)) -- | Tier levels representing the risk and side-effects of tool execution. data ToolTier @@ -278,6 +280,26 @@ Nothing Nothing +checkPolicyApproval :: + MonadIO m => + ToolTier -> + SecurityPolicy -> + Text -> + Text -> + Value -> + m (Either LangchainError ()) +checkPolicyApproval tier policy tName tDesc args = + case checkTierPermission policy tier of + Left err -> pure $ Left err + Right () -> do + mApproval <- case policyApprovalHook policy of + Nothing -> pure Approved + Just hook -> liftIO $ hook (ApprovalRequest tName tier args tDesc) + case mApproval of + Denied reason -> + pure (Left (toolError ("Execution rejected by approval policy: " <> reason) (Just tName) Nothing)) + Approved -> pure $ Right () + -- | Wrap a 'Tool m' with permission tier enforcement, sandboxing, and optional HITL approval. withSecurity :: MonadIO m => @@ -288,17 +310,27 @@ withSecurity tier policy tool@Tool {..} = tool { toolExecute = \args -> do - case checkTierPermission policy tier of + eOk <- checkPolicyApproval tier policy toolName toolDescription args + case eOk of Left err -> pure $ Left err - Right () -> do - -- Check HITL approval if hook is defined - mApproval <- case policyApprovalHook policy of - Nothing -> pure Approved - Just hook -> liftIO $ hook (ApprovalRequest toolName tier args toolDescription) - case mApproval of - Denied reason -> - pure $ - Left $ - toolError ("Execution rejected by approval policy: " <> reason) (Just toolName) Nothing - Approved -> toolExecute args + Right () -> toolExecute args } + +-- | Wrap a 'StreamingTool m' with permission tier enforcement, sandboxing, and optional HITL approval. +withStreamingSecurity :: + MonadIO m => + ToolTier -> + SecurityPolicy -> + StreamingTool m -> + StreamingTool m +withStreamingSecurity tier policy st@StreamingTool {..} = + let securedDef = withSecurity tier policy streamingToolDef + in st + { streamingToolDef = securedDef + , streamingToolExecute = \args -> do + eOk <- + lift $ checkPolicyApproval tier policy (toolName securedDef) (toolDescription securedDef) args + case eOk of + Left err -> pure $ Left err + Right () -> streamingToolExecute args + } diff --git a/src/Langchain/Tool/Shell.hs b/src/Langchain/Tool/Shell.hs --- a/src/Langchain/Tool/Shell.hs +++ b/src/Langchain/Tool/Shell.hs @@ -1,5 +1,6 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TypeApplications #-} {- | Module : Langchain.Tool.Shell @@ -15,27 +16,96 @@ module Langchain.Tool.Shell ( shellTool , secureShellTool + , streamingShellTool + , streamingShellToolWithCancel ) where -import Control.Exception (SomeException, try) +import Control.Concurrent.Async (AsyncCancelled (..), async, cancel, concurrently, wait) +import Control.Concurrent.STM + ( TBQueue + , atomically + , newEmptyTMVarIO + , newTBQueueIO + , orElse + , readTBQueue + , readTMVar + , tryPutTMVar + , writeTBQueue + ) +import Control.Exception (SomeException, catch, finally, fromException, onException, throwIO, try) +import Control.Monad (void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value (..), object, (.=)) import Data.Aeson.Types (parseEither, (.:?)) +import Data.Conduit (ConduitT, yield) +import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef) import Data.Text (Text) import qualified Data.Text as T +import qualified Data.Text.IO as TIO import System.Exit (ExitCode (..)) -import System.Process (readProcessWithExitCode) +import System.IO (BufferMode (LineBuffering), Handle, hClose, hIsEOF, hSetBuffering) +import System.Process + ( CreateProcess (..) + , StdStream (..) + , createProcess + , interruptProcessGroupOf + , proc + , readProcessWithExitCode + , terminateProcess + , waitForProcess + ) -import Langchain.Error (toolError) -import Langchain.Tool (Tool (..), createTool) +import Langchain.Error (LangchainError, toolError) +import Langchain.Stream (CancelToken, isCancelled, waitCancel) +import Langchain.Tool (StreamingTool, Tool (..), createStreamingTool, createTool) import Langchain.Tool.Security ( SecurityPolicy , ToolTier (..) , defaultSecurityPolicy , validateCommandSafety , withSecurity + , withStreamingSecurity ) +shellToolDescription :: Text +shellToolDescription = + "Execute a shell command line (e.g. bash/sh) and return stdout and stderr output." + +shellToolSchema :: Value +shellToolSchema = + object + [ "type" .= ("object" :: Text) + , "properties" + .= object + [ "command" + .= object + [ "type" .= ("string" :: Text) + , "description" .= ("The shell command line to execute" :: Text) + ] + ] + , "required" .= (["command"] :: [Text]) + ] + +extractShellCommand :: Value -> Either LangchainError Text +extractShellCommand = \case + Object o -> case parseEither (.:? "command") o of + Right (Just cmd) -> Right cmd + _ -> Left $ toolError "Missing 'command' parameter" (Just "shell_command") Nothing + _ -> Left $ toolError "Invalid arguments object" (Just "shell_command") Nothing + +formatShellResult :: ExitCode -> Text -> Text -> Text +formatShellResult ExitSuccess stdoutTxt stderrTxt = + let out = T.strip stdoutTxt + err = T.strip stderrTxt + in if T.null out + then if T.null err then "Command completed with no output." else err + else out +formatShellResult (ExitFailure code) stdoutTxt stderrTxt = + let combined = T.strip (stdoutTxt <> "\n" <> stderrTxt) + in "Command exited with code " + <> T.pack (show code) + <> (if T.null combined then "" else ": " <> combined) + -- | Tool that executes a shell command with default security policy. shellTool :: MonadIO m => Tool m shellTool = secureShellTool defaultSecurityPolicy @@ -46,42 +116,158 @@ withSecurity TierExecution policy $ createTool "shell_command" - "Execute a shell command line (e.g. bash/sh) and return stdout and stderr output." - ( object - [ "type" .= ("object" :: Text) - , "properties" - .= object - [ "command" - .= object - [ "type" .= ("string" :: Text) - , "description" .= ("The shell command line to execute" :: Text) - ] - ] - , "required" .= (["command"] :: [Text]) - ] + shellToolDescription + shellToolSchema + ( \args -> case extractShellCommand args of + Left err -> pure $ Left err + Right cmd -> case validateCommandSafety policy cmd of + Left err -> pure $ Left err + Right () -> do + eRes <- liftIO $ try (readProcessWithExitCode "sh" ["-c", T.unpack cmd] "") + case eRes of + Left err -> + pure $ Left $ toolError (T.pack $ show (err :: SomeException)) (Just "shell_command") Nothing + Right (code, stdoutStr, stderrStr) -> + pure $ Right $ formatShellResult code (T.pack stdoutStr) (T.pack stderrStr) ) - ( \case - Object o -> case parseEither (.:? "command") o of - Right (Just cmd) -> case validateCommandSafety policy cmd of - Left err -> pure $ Left err - Right () -> do - eRes <- liftIO $ try (readProcessWithExitCode "sh" ["-c", T.unpack cmd] "") - case eRes of - Left err -> - pure $ Left $ toolError (T.pack $ show (err :: SomeException)) (Just "shell_command") Nothing - Right (ExitSuccess, stdoutStr, stderrStr) -> - let out = T.strip (T.pack stdoutStr) - err = T.strip (T.pack stderrStr) - in if T.null out - then if T.null err then pure $ Right "Command completed with no output." else pure $ Right err - else pure $ Right out - Right (ExitFailure code, stdoutStr, stderrStr) -> - let combined = T.strip (T.pack (stdoutStr <> "\n" <> stderrStr)) - in pure $ - Right $ - "Command exited with code " - <> T.pack (show code) - <> (if T.null combined then "" else ": " <> combined) - _ -> pure $ Left $ toolError "Missing 'command' parameter" (Just "shell_command") Nothing - _ -> pure $ Left $ toolError "Invalid arguments object" (Just "shell_command") Nothing + +-- | Streaming shell tool that yields stdout and stderr lines incrementally while enforcing 'SecurityPolicy'. +streamingShellTool :: MonadIO m => SecurityPolicy -> StreamingTool m +streamingShellTool = makeStreamingShellTool Nothing + +{- | Streaming shell tool that monitors a 'CancelToken' and terminates the +spawned shell child process immediately on cancellation. +-} +streamingShellToolWithCancel :: MonadIO m => CancelToken -> SecurityPolicy -> StreamingTool m +streamingShellToolWithCancel tok = makeStreamingShellTool (Just tok) + +makeStreamingShellTool :: MonadIO m => Maybe CancelToken -> SecurityPolicy -> StreamingTool m +makeStreamingShellTool mbTok policy = + withStreamingSecurity TierExecution policy $ + createStreamingTool + "shell_command" + shellToolDescription + shellToolSchema + ( \args -> case extractShellCommand args of + Left err -> pure $ Left err + Right cmd -> case validateCommandSafety policy cmd of + Left err -> pure $ Left err + Right () -> streamShellCommand mbTok cmd ) + +streamShellCommand :: + MonadIO m => + Maybe CancelToken -> + Text -> + ConduitT () Text m (Either LangchainError Text) +streamShellCommand mbTok cmd = do + cancelledNow <- maybe (pure False) isCancelled mbTok + if cancelledNow + then pure $ Left $ toolError "Shell command cancelled" (Just "shell_command") Nothing + else startShellProcess mbTok cmd + +startShellProcess :: + MonadIO m => + Maybe CancelToken -> + Text -> + ConduitT () Text m (Either LangchainError Text) +startShellProcess mbTok cmd = do + eProc <- + liftIO $ + try $ + createProcess + (proc "sh" ["-c", T.unpack cmd]) + { std_in = NoStream + , std_out = CreatePipe + , std_err = CreatePipe + , create_group = True + } + case eProc of + Left ex -> + pure $ Left $ toolError (T.pack $ show (ex :: SomeException)) (Just "shell_command") Nothing + Right (_, Just hOut, Just hErr, ph) -> do + (queue, doneVar, worker) <- liftIO $ do + hSetBuffering hOut LineBuffering + hSetBuffering hErr LineBuffering + q <- newTBQueueIO 128 + done <- newEmptyTMVarIO + outRef <- newIORef ([] :: [Text]) + errRef <- newIORef ([] :: [Text]) + resRef <- newIORef (Right "" :: Either LangchainError Text) + w <- + async $ + ( do + _ <- concurrently (pumpHandle hOut q outRef) (pumpHandle hErr q errRef) + code <- waitForProcess ph + outChunks <- T.concat . reverse <$> readIORef outRef + errChunks <- T.concat . reverse <$> readIORef errRef + writeIORef resRef (Right $ formatShellResult code outChunks errChunks) + ) + `catch` ( \ex -> case fromException ex of + Just AsyncCancelled -> do + void (try @SomeException (interruptProcessGroupOf ph)) + void (try @SomeException (terminateProcess ph)) + void (try @SomeException (waitForProcess ph)) + throwIO (ex :: SomeException) + Nothing -> + writeIORef + resRef + (Left $ toolError (T.pack $ show (ex :: SomeException)) (Just "shell_command") Nothing) + ) + `finally` ( do + void (try @SomeException (hClose hOut)) + void (try @SomeException (hClose hErr)) + readIORef resRef >>= atomically . void . tryPutTMVar done + ) + pure (q, done, w) + drainLoop queue doneVar worker + Right (_, mbOut, mbErr, ph) -> do + liftIO $ do + mapM_ (try @SomeException . hClose) mbOut + mapM_ (try @SomeException . hClose) mbErr + void (try @SomeException (interruptProcessGroupOf ph)) + void (try @SomeException (terminateProcess ph)) + pure $ + Left $ + toolError "Failed to open stdout/stderr pipes for shell_command" (Just "shell_command") Nothing + where + pumpHandle :: Handle -> TBQueue Text -> IORef [Text] -> IO () + pumpHandle h q ref = loop + where + loop = do + eof <- hIsEOF h + if eof + then pure () + else do + line <- TIO.hGetLine h + let chunk = line <> "\n" + modifyIORef' ref (chunk :) + atomically $ writeTBQueue q chunk + loop + + drainLoop queue doneVar worker = loop + where + loop = do + next <- + liftIO $ + atomically + ( case mbTok of + Just tok -> + (Left Nothing <$ waitCancel tok) + `orElse` (Right <$> readTBQueue queue) + `orElse` (Left . Just <$> readTMVar doneVar) + Nothing -> + (Right <$> readTBQueue queue) + `orElse` (Left . Just <$> readTMVar doneVar) + ) + `onException` cancel worker + case next of + Right chunk -> do + yield chunk + loop + Left Nothing -> do + liftIO $ cancel worker + pure $ Left $ toolError "Shell command cancelled" (Just "shell_command") Nothing + Left (Just res) -> do + void $ liftIO $ wait worker + pure res diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -3,6 +3,7 @@ import Test.Tasty import qualified Test.Langchain.Callback as CallbackTest +import qualified Test.Langchain.Eval as EvalTest import qualified Test.Langchain.Model as ModelTest import qualified Test.Langchain.Monad as MonadTest import qualified Test.Langchain.Observability as ObservabilityTest @@ -30,4 +31,5 @@ , CallbackTest.tests , RetrieverTest.tests , ResilienceTest.tests + , EvalTest.tests ] diff --git a/test/Test/Langchain/Eval.hs b/test/Test/Langchain/Eval.hs new file mode 100644 --- /dev/null +++ b/test/Test/Langchain/Eval.hs @@ -0,0 +1,109 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Test.Langchain.Eval (tests) where + +import Control.Monad.Except (runExceptT) +import qualified Data.Aeson as Aeson +import Data.Text (Text) +import Test.Tasty +import Test.Tasty.HUnit + +import Langchain.Agent.ReAct (AgentStep (..), AgentTrace (..)) +import Langchain.Error (LangchainError) +import Langchain.Eval +import Langchain.Model.Types (Role (..), ToolCall (..), assistantMessage, textMessage) +import Test.Langchain.TestModel (TestChatModel (..)) + +tests :: TestTree +tests = + testGroup + "Langchain.Eval" + [ testCase "runEvalSuite computes pass rate, mean score, and latency across mixed cases" $ do + let cases :: [EvalCase Text Text] + cases = + [ EvalCase "query-1" (Just "answer-1") Nothing + , EvalCase "query-2" (Just "answer-2") Nothing + ] + mockRunner :: Text -> IO (Either LangchainError Text) + mockRunner "query-1" = pure $ Right "answer-1" + mockRunner "query-2" = pure $ Right "wrong-answer" + mockRunner _ = pure $ Right "" + report <- runEvalSuite mockRunner [exactMatchScorer] cases + reportTotalCases report @?= 2 + reportPassedCases report @?= 1 + reportPassRate report @?= 0.5 + reportMeanScore report @?= 0.5 + assertBool "Non-negative latency" (reportMeanLatencyMicros report >= 0) + length (reportCases report) @?= 2 + , testCase "containsScorer verifies substring presence" $ do + let cPass :: EvalCase Text Text + cPass = EvalCase "in" (Just "needle") Nothing + cFail :: EvalCase Text Text + cFail = EvalCase "in" (Just "missing") Nothing + scPass <- scoreCase containsScorer cPass ("haystack needle haystack" :: Text) + scorePassed scPass @?= True + scoreValue scPass @?= 1.0 + scFail <- scoreCase containsScorer cFail ("haystack other haystack" :: Text) + scorePassed scFail @?= False + scoreValue scFail @?= 0.0 + , testCase "toolCallExactScorer verifies sequence of invoked tool names in AgentTrace" $ do + let tc1 = ToolCall "c1" "function" "search" (Aeson.object []) + tc2 = ToolCall "c2" "function" "calculator" (Aeson.object []) + action1 = AgentAction (assistantMessage "") [tc1] + action2 = AgentAction (assistantMessage "") [tc2] + finish = AgentFinish (assistantMessage "Done") + obs = textMessage Tool "obs" + trace = + AgentTrace + { traceSteps = [(action1, [obs]), (action2, [obs]), (finish, [])] + , traceFinalAnswer = assistantMessage "Done" + , traceIterations = 3 + } + matchingScorer = toolCallExactScorer ["search", "calculator"] + mismatchScorer = toolCallExactScorer ["calculator", "search"] + evalCase = EvalCase ("test" :: Text) Nothing Nothing + resMatch <- scoreCase matchingScorer evalCase trace + scorePassed resMatch @?= True + scoreValue resMatch @?= 1.0 + resMismatch <- scoreCase mismatchScorer evalCase trace + scorePassed resMismatch @?= False + scoreValue resMismatch @?= 0.0 + , testCase "llmJudgeScorer parses JSON judge evaluation and score" $ do + let judgeModel = + TestChatModel + "{\"passed\": true, \"score\": 0.95, \"reason\": \"Accurate and concise.\"}" + "judge-mock" + scorer = llmJudgeScorer judgeModel "Evaluate answer accuracy" + evalCase :: EvalCase Text Text + evalCase = EvalCase "What is 2+2?" (Just "4") Nothing + eScore <- runExceptT $ scoreCase scorer evalCase "4" + case eScore of + Left err -> assertFailure ("Judge scorer failed: " ++ show err) + Right score -> do + scorePassed score @?= True + scoreValue score @?= 0.95 + scoreReason score @?= Just "Accurate and concise." + , testCase "EvalReport round-trips cleanly via Aeson encode and decode" $ do + let score = EvalScore 1.0 True (Just "Matched perfectly") + caseRes = + CaseResult + { caseInput = Aeson.String "input-1" + , caseExpected = Just (Aeson.String "expected-1") + , caseActual = Right (Aeson.String "actual-1") + , caseScores = [CaseScoreResult "exact_match" score] + , caseLatencyMicros = 1200 + , casePassed = True + } + report = + EvalReport + { reportTotalCases = 1 + , reportPassedCases = 1 + , reportPassRate = 1.0 + , reportMeanScore = 1.0 + , reportMeanLatencyMicros = 1200 + , reportCases = [caseRes] + } + encoded = Aeson.encode report + decoded = Aeson.decode encoded + decoded @?= Just report + ] diff --git a/test/Test/Langchain/Stream.hs b/test/Test/Langchain/Stream.hs --- a/test/Test/Langchain/Stream.hs +++ b/test/Test/Langchain/Stream.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} module Test.Langchain.Stream (tests) where @@ -5,10 +6,21 @@ import Test.Tasty import Test.Tasty.HUnit -import Control.Monad.Except (runExceptT) +import Control.Concurrent (threadDelay) +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, tryTakeMVar) +import Control.Exception (finally) +import Control.Monad.Except (runExceptT, throwError) +import Control.Monad.IO.Class (liftIO) import Control.Monad.Trans.Resource (runResourceT) +import Data.Aeson (object, (.=)) +import Data.Conduit (runConduit, yield, (.|)) +import qualified Data.Conduit.List as CL +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.Time.Clock (diffUTCTime, getCurrentTime) +import Langchain.Error (llmError) import Langchain.Model +import Langchain.Monad (runLangchainT) import Langchain.Stream import Test.Langchain.TestModel (TestChatModel (..)) @@ -30,4 +42,94 @@ chunkText c @?= "Streamed content" extractMessageText (finalMessage e) @?= "Streamed content" _ -> assertFailure $ "Unexpected event sequence: " ++ show events + , testCase "liftChatStream forwards events into LangchainT () IO" $ do + let model = TestChatModel "Bridged stream" "mock-gpt" + input = [userMessage "Bridge test"] + res <- runLangchainT () $ collectEvents $ liftChatStream (stream model input Nothing) + case res of + Left err -> assertFailure $ "Unexpected error: " ++ show err + Right events -> do + length events @?= 3 + case reverse events of + LLMEnd _ msg _ : _ -> extractMessageText msg @?= "Bridged stream" + _ -> assertFailure $ "Expected LLMEnd, got: " ++ show events + , testCase "liftChatStream propagates LangchainError mid-stream" $ do + let errStream :: ChatStream + errStream = do + yield $ LLMStart "r1" "mock" [userMessage "Hi"] + yield $ LLMChunk "r1" "partial" Nothing + throwError $ llmError "mid-stream boom" Nothing Nothing + res <- runExceptT $ collectEvents $ liftChatStream errStream + case res of + Left _ -> pure () + Right events -> assertFailure $ "Expected failure, got: " ++ show events + , testCase "runStreamWith invokes callback and resets across LLMStart turns" $ do + let tc = ToolCall "call_1" "function" "calc" (object ["x" .= (1 :: Int)]) + multiTurnStream :: EventStream IO + multiTurnStream = do + yield $ LLMStart "r1" "mock" [] + yield $ LLMChunk "r1" "Calling tool..." Nothing + yield $ LLMChunk "r1" "" (Just tc) + yield $ LLMEnd "r1" ((assistantMessage "Calling tool...") {messageToolCalls = Just [tc]}) Nothing + yield $ ToolStart "r1" "calc" (object ["x" .= (1 :: Int)]) + yield $ ToolEnd "r1" "calc" (object ["res" .= (2 :: Int)]) + yield $ LLMStart "r2" "mock" [] + yield $ LLMChunk "r2" "Final " Nothing + yield $ LLMChunk "r2" "answer" Nothing + seenRef <- newIORef ([] :: [StreamEvent]) + finalMsg <- runStreamWith (\ev -> liftIO $ modifyIORef' seenRef (ev :)) multiTurnStream + seen <- readIORef seenRef + length seen @?= 9 + extractMessageText finalMsg @?= "Final answer" + messageToolCalls finalMsg @?= Nothing + , testCase "liftChatStreamResource cancels upstream within 200ms on take 1" $ do + cleanedUp <- newEmptyMVar + let slowUpstream :: ChatStream + slowUpstream = do + yield $ LLMStart "slow" "mock" [] + liftIO $ + threadDelay 5000000 + `finally` putMVar cleanedUp () + yield $ LLMChunk "slow" "late" Nothing + t0 <- getCurrentTime + res <- + runResourceT $ + runExceptT $ + runConduit $ + liftChatStreamResource slowUpstream .| CL.take 1 + t1 <- getCurrentTime + case res of + Left err -> assertFailure $ "Unexpected error: " ++ show err + Right evs -> length evs @?= 1 + threadDelay 50000 + mbDone <- tryTakeMVar cleanedUp + mbDone @?= Just () + let elapsed = realToFrac (diffUTCTime t1 t0) :: Double + assertBool ("Expected cancellation < 0.2s, got: " ++ show elapsed) (elapsed < 0.2) + , testCase "liftChatStreamWithCancel stops on CancelToken and keeps partial text" $ do + tok <- newCancelToken + cleanedUp <- newEmptyMVar + gate <- newEmptyMVar :: IO (MVar ()) + let partialUpstream :: ChatStream + partialUpstream = do + yield $ LLMStart "c1" "mock" [] + yield $ LLMChunk "c1" "partial-" Nothing + liftIO $ + takeMVar gate + `finally` putMVar cleanedUp () + yield $ LLMChunk "c1" "never" Nothing + res <- + runLangchainT () $ + runStreamWith + ( \case + LLMChunk {} -> triggerCancel tok + _ -> pure () + ) + (liftChatStreamWithCancel tok partialUpstream) + case res of + Left err -> assertFailure $ "Unexpected error: " ++ show err + Right msg -> extractMessageText msg @?= "partial-" + threadDelay 50000 + mbDone <- tryTakeMVar cleanedUp + mbDone @?= Just () ] diff --git a/test/Test/Langchain/Tool.hs b/test/Test/Langchain/Tool.hs --- a/test/Test/Langchain/Tool.hs +++ b/test/Test/Langchain/Tool.hs @@ -3,11 +3,25 @@ module Test.Langchain.Tool (tests) where +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (async, cancel, wait) import Data.Aeson import Data.Aeson.Types (parseEither) +import Data.Conduit (fuseUpstream, runConduit, yield) +import qualified Data.Conduit.List as CL +import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Text (Text) +import qualified Data.Text as T +import Data.Time.Clock (diffUTCTime, getCurrentTime) import Langchain.Error +import Langchain.Stream + ( StreamEvent (ToolChunk, ToolEnd, ToolErrorEvent, ToolStart) + , newCancelToken + , triggerCancel + ) import Langchain.Tool +import Langchain.Tool.Security (readOnlySecurityPolicy, unrestrictedSecurityPolicy) +import Langchain.Tool.Shell (streamingShellTool, streamingShellToolWithCancel) import Test.Tasty import Test.Tasty.HUnit @@ -36,4 +50,104 @@ let t = createTool "calculator" "Performs math" (object []) calcExec res <- toolExecute t (object []) res @?= Right "42" + , testCase "streamToolCall emits ToolStart and ToolEnd on success" $ do + let t = createTool "calculator" "Performs math" (object []) calcExec + eventsRef <- newIORef ([] :: [StreamEvent]) + res <- + runConduit $ + streamToolCall "run-1" t (object []) + `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev])) + res @?= Right "42" + events <- readIORef eventsRef + events + @?= [ ToolStart "run-1" "calculator" (object []) + , ToolEnd "run-1" "calculator" (toJSON ("42" :: Text)) + ] + , testCase "streamToolCall emits ToolStart and ToolErrorEvent on failure" $ do + let t = createTool "calculator" "Performs math" (object []) calcExec + eventsRef <- newIORef ([] :: [StreamEvent]) + res <- + runConduit $ + streamToolCall "run-2" t (Number 1) + `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev])) + case res of + Left _ -> pure () + Right out -> assertFailure $ "Expected error, got: " ++ show out + events <- readIORef eventsRef + case events of + [ToolStart "run-2" "calculator" (Number 1), ToolErrorEvent "run-2" "calculator" _] -> pure () + _ -> assertFailure $ "Unexpected events: " ++ show events + , testCase "streamStreamingToolCall emits incremental ToolChunk events" $ do + let sTool = + createStreamingTool "streamer" "Streams chunks" (object []) $ \_ -> do + yield "chunk-1 " + yield "chunk-2" + pure (Right "chunk-1 chunk-2") + eventsRef <- newIORef ([] :: [StreamEvent]) + res <- + runConduit $ + streamStreamingToolCall "run-3" sTool (object []) + `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev])) + res @?= Right "chunk-1 chunk-2" + events <- readIORef eventsRef + events + @?= [ ToolStart "run-3" "streamer" (object []) + , ToolChunk "run-3" "streamer" "chunk-1 " + , ToolChunk "run-3" "streamer" "chunk-2" + , ToolEnd "run-3" "streamer" (toJSON ("chunk-1 chunk-2" :: Text)) + ] + , testCase "streamingShellTool streams stdout lines and enforces SecurityPolicy" $ do + let sShell = streamingShellTool unrestrictedSecurityPolicy + eventsRef <- newIORef ([] :: [StreamEvent]) + res <- + runConduit $ + streamStreamingToolCall + "run-sh" + sShell + (object ["command" .= ("printf 'line1\\nline2\\n'" :: Text)]) + `fuseUpstream` CL.mapM_ (\ev -> modifyIORef' eventsRef (++ [ev])) + res @?= Right "line1\nline2" + events <- readIORef eventsRef + let chunks = [txt | ToolChunk _ _ txt <- events] + chunks @?= ["line1\n", "line2\n"] + + let roShell = streamingShellTool readOnlySecurityPolicy + roRes <- toolExecute (streamingToolDef roShell) (object ["command" .= ("echo hi" :: Text)]) + case roRes of + Left err -> assertBool "Mentions permission" ("permission" `T.isInfixOf` T.toLower (errorMessage err)) + Right ok -> assertFailure $ "Expected security denial, got: " ++ show ok + , testCase "cancelling streamingShellTool via Async.cancel terminates sleep child promptly" $ do + let sShell = streamingShellTool unrestrictedSecurityPolicy + t0 <- getCurrentTime + a <- + async $ + runConduit $ + streamStreamingToolCall + "run-cancel-async" + sShell + (object ["command" .= ("sleep 10" :: Text)]) + `fuseUpstream` CL.sinkNull + threadDelay 100000 + cancel a + t1 <- getCurrentTime + let elapsed = realToFrac (diffUTCTime t1 t0) :: Double + assertBool ("Expected prompt Async cancellation (< 2s), got: " ++ show elapsed) (elapsed < 2.0) + , testCase "cancelling streamingShellToolWithCancel via CancelToken terminates promptly" $ do + tok <- newCancelToken + let sShell = streamingShellToolWithCancel tok unrestrictedSecurityPolicy + t0 <- getCurrentTime + a <- + async $ + runConduit $ + streamStreamingToolCall + "run-cancel-tok" + sShell + (object ["command" .= ("sleep 10" :: Text)]) + `fuseUpstream` CL.sinkNull + threadDelay 100000 + triggerCancel tok + _ <- wait a + t1 <- getCurrentTime + let elapsed = realToFrac (diffUTCTime t1 t0) :: Double + assertBool ("Expected prompt CancelToken termination (< 2s), got: " ++ show elapsed) (elapsed < 2.0) ]