[agents] Add Agent to call LangServe (langserve-invoke) (LangStream#673)

examples/applications/langserve-invoke/README.md

@@ -0,0 +1,68 @@
+# Invoking a LangServe service
+
+This sample application explains how to invoke a LangServe service and leverage streaming capabilities.
+
+## Set up your LangServe environment
+
+Start you LangServe application, the example below is using the LangServe sample [application](https://github.com/langchain-ai/langserve)
+
+```python
+#!/usr/bin/env python
+from fastapi import FastAPI
+from langchain.prompts import ChatPromptTemplate
+from langchain.chat_models import ChatAnthropic, ChatOpenAI
+from langserve import add_routes
+
+
+app = FastAPI(
+  title="LangChain Server",
+  version="1.0",
+  description="A simple api server using Langchain's Runnable interfaces",
+)
+
+model = ChatOpenAI()
+prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
+add_routes(
+    app,
+    prompt | model,
+    path="/chain",
+)
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="localhost", port=8000)
+```
+
+## Configure you OpenAI API Key and run the application
+
+```bash
+export OPENAI_API_KEY=...
+pip install fastapi langserve langchain openai sse_starlette uvicorn
+python example.py
+```
+
+The sample application is exposing a chain at http://localhost:8000/chain/stream and http://localhost:8000/chain/invoke.
+
+The application, running in docker, connects to http://host.docker.internal:8000/chain/stream
+
+LangStream sends an input like this:
+
+```json
+{
+    "input": {
+        "topic": "cats"
+    }
+}
+```
+
+When "topic" is the topic of the joke you want to generate and it is taken from the user input.
+
+## Deploy the LangStream application
+```
+./bin/langstream docker run test -app examples/applications/langserve-invoke
+```
+
+## Interact with the application
+
+You can now interact with the application using the UI opening your browser at http://localhost:8092/

examples/applications/langserve-invoke/example.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from fastapi import FastAPI
+from langchain.prompts import ChatPromptTemplate
+from langchain.chat_models import ChatOpenAI
+from langserve import add_routes
+
+
+app = FastAPI(
+    title="LangChain Server",
+    version="1.0",
+    description="A simple api server using Langchain's Runnable interfaces",
+)
+
+model = ChatOpenAI()
+prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
+add_routes(
+    app,
+    prompt | model,
+    path="/chain",
+    )
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="localhost", port=8000)

examples/applications/langserve-invoke/gateways.yaml

@@ -0,0 +1,25 @@
+#
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+gateways:
+  - id: chat
+    type: chat
+    chat-options:
+      answers-topic: streaming-answers-topic
+      questions-topic: input-topic
+      headers:
+        - value-from-parameters: session-id

examples/applications/langserve-invoke/pipeline.yaml

@@ -0,0 +1,41 @@
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+topics:
+  - name: "input-topic"
+    creation-mode: create-if-not-exists
+  - name: "output-topic"
+    creation-mode: create-if-not-exists
+  - name: "streaming-answers-topic"
+    creation-mode: create-if-not-exists
+pipeline:
+  - type: "langserve-invoke"
+    input: input-topic
+    output: output-topic
+    id: step1
+    configuration:
+      output-field: value.answer
+      stream-to-topic: streaming-answers-topic
+      stream-response-field: value
+      min-chunks-per-message: 10
+      debug: false
+      method: POST
+      allow-redirects: true
+      handle-cookies: false
+      url: "http://host.docker.internal:8000/chain/stream"
+      fields:
+        - name: topic
+          expression: "value"

examples/applications/langserve/README.md → examples/applications/python/langserve-service/README.md

@@ -10,7 +10,7 @@ Export to the ENV the access key to OpenAI
 export OPEN_AI_ACCESS_KEY=...
 ```
 
-The default [secrets file](../../secrets/secrets.yaml) reads from the ENV. Check out the file to learn more about
+The default [secrets file](../../../secrets/secrets.yaml) reads from the ENV. Check out the file to learn more about
 the default settings, you can change them by exporting other ENV variables.
 
 
@@ -26,7 +26,7 @@ export LANGSMITH_APIKEY=xxxxx
 ## Deploy the LangStream application
 
 ```
-./bin/langstream docker run test -app examples/applications/langserve -s examples/secrets/secrets.yaml --start-broker=false
+./bin/langstream docker run test -app examples/applications/python/langserve-service -s examples/secrets/secrets.yaml --start-broker=false
 ```
 
 ## Interact with the application

langstream-agents/langstream-agent-azure-blob-storage-source/src/main/java/ai/langstream/agents/azureblobstorage/source/AzureBlobStorageSource.java → langstream-agents/langstream-agent-azure-blob-storage-source/src/main/java/ai/langstream/agents/azureblobstorage/AzureBlobStorageSource.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.agents.azureblobstorage.source;
+package ai.langstream.agents.azureblobstorage;
 
 import ai.langstream.api.runner.code.AbstractAgentCode;
 import ai.langstream.api.runner.code.AgentSource;

langstream-agents/langstream-agent-azure-blob-storage-source/src/main/java/ai/langstream/agents/azureblobstorage/source/AzureBlobStorageSourceCodeProvider.java → langstream-agents/langstream-agent-azure-blob-storage-source/src/main/java/ai/langstream/agents/azureblobstorage/AzureBlobStorageSourceCodeProvider.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.agents.azureblobstorage.source;
+package ai.langstream.agents.azureblobstorage;
 
 import ai.langstream.api.runner.code.AgentCode;
 import ai.langstream.api.runner.code.AgentCodeProvider;

langstream-agents/langstream-agent-azure-blob-storage-source/src/main/resources/META-INF/services/ai.langstream.api.runner.code.AgentCodeProvider

@@ -1 +1 @@
-ai.langstream.agents.azureblobstorage.source.AzureBlobStorageSourceCodeProvider
+ai.langstream.agents.azureblobstorage.AzureBlobStorageSourceCodeProvider

langstream-agents/langstream-agent-azure-blob-storage-source/src/test/java/ai/langstream/agents/azureblobstorage/source/AzureBlobStorageSourceTest.java → langstream-agents/langstream-agent-azure-blob-storage-source/src/test/java/ai/langstream/agents/azureblobstorage/AzureBlobStorageSourceTest.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.agents.azureblobstorage.source;
+package ai.langstream.agents.azureblobstorage;
 
 import static org.junit.jupiter.api.Assertions.*;
 

langstream-agents/langstream-agent-http-request/pom.xml

@@ -65,6 +65,11 @@
       <artifactId>junit-jupiter</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.github.tomakehurst</groupId>
+      <artifactId>wiremock</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgent.java → langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/http/HttpRequestAgent.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.agents.azureblobstorage.source;
+package ai.langstream.agents.http;
 
 import ai.langstream.ai.agents.commons.JsonRecord;
 import ai.langstream.ai.agents.commons.MutableRecord;
@@ -191,6 +191,12 @@ public void processRecord(Record record, RecordSink recordSink) {
                                     recordSink.emit(
                                             new SourceRecordAndResult(record, List.of(), e));
                                 }
+                            })
+                    .exceptionally(
+                            error -> {
+                                log.error("Error processing record: {}", record, error);
+                                recordSink.emit(new SourceRecordAndResult(record, null, error));
+                                return null;
                             });
         } catch (Throwable error) {
             log.error("Error processing record: {}", record, error);

langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgentProvider.java → langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/http/HttpRequestAgentProvider.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.agents.azureblobstorage.source;
+package ai.langstream.agents.http;
 
 import ai.langstream.api.runner.code.AgentCodeProvider;
 import ai.langstream.api.runner.code.AgentProcessor;
@@ -25,7 +25,11 @@
 public class HttpRequestAgentProvider implements AgentCodeProvider {
 
     private static final Map<String, Supplier<AgentProcessor>> FACTORIES =
-            Map.of("http-request", HttpRequestAgent::new);
+            Map.of(
+                    "http-request",
+                    HttpRequestAgent::new,
+                    "langserve-invoke",
+                    LangServeInvokeAgent::new);
 
     @Override
     public boolean supports(String agentType) {