xintongsong commented on code in PR #126:
URL: https://github.com/apache/flink-agents/pull/126#discussion_r2318356303
##########
python/flink_agents/api/events/tool_event.py:
##########
@@ -42,9 +43,9 @@ class ToolResponseEvent(Event):
----------
request : ToolRequestEvent
The correspond request of the response.
- response : Any
- The response from the tool.
+ responses : Dict[UUID, Any]
Review Comment:
Why do we need `ToolRequestEvent` in `ToolResponseEvent`? Can we replace it
with the request id, like what we do for `ChatResponseEvent`?
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
Review Comment:
What is this `request_id`? IIUC, this is the id of the initial
`ChatRequestEvent`? We should make that explicit, perhaps with the name?
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
+ model: str,
+ chat_model: BaseChatModelSetup,
+ messages: List[ChatMessage],
+ short_term_memory: MemoryObject,
+ ctx: RunnerContext,
Review Comment:
Why do we need `short_term_memory` as an argument, while we already have
`ctx`?
##########
plan/src/main/java/org/apache/flink/agents/plan/Action.java:
##########
@@ -58,4 +68,8 @@ public Function getExec() {
public List<String> getListenEventTypes() {
return listenEventTypes;
}
+
+ public Map<String, Object> getParams() {
Review Comment:
I'd suggest the name `getActionConfig`.
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
+ model: str,
+ chat_model: BaseChatModelSetup,
Review Comment:
Seems `chat_model` is converted from `model`. Why not do it inside `chat()`?
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
+ model: str,
+ chat_model: BaseChatModelSetup,
+ messages: List[ChatMessage],
+ short_term_memory: MemoryObject,
+ ctx: RunnerContext,
+) -> None:
+ """Chat with llm.
+
+ If there is no tool call generated, we return the chat response event
directly,
+ otherwise, we generate tool request event according to the tool calls in
chat model
+ response, and save the request and response messages in tool call context.
+ """
+ # TODO: support async execution of chat.
+ response = chat_model.chat(messages)
+
+ # generate tool request event according tool calls in response
+ if len(response.tool_calls) > 0:
+ # TODO: Because memory doesn't support remove currently, so we use
+ # dict to store tool context in memory and remove the specific
+ # tool context from dict after consuming. This will cause write and
+ # read amplification for we need get the whole dict and overwrite it
+ # to memory each time we update a specific tool context.
+ # After memory supports remove, we can use
"TOOL_CALL_CONTEXT/request_id"
+ # to store and remove the specific tool context directly.
+
+ # get tool call context
+ tool_call_context = short_term_memory.get(TOOL_CALL_CONTEXT)
+ if not tool_call_context:
+ tool_call_context = {}
+ if request_id not in tool_call_context:
+ tool_call_context[request_id] = copy.deepcopy(messages)
+ # append response to tool call context
+ tool_call_context[request_id].append(response)
+ # update tool call context
+ short_term_memory.set(TOOL_CALL_CONTEXT, tool_call_context)
+ ctx.send_event(
+ ToolRequestEvent(
+ id=request_id,
+ model=model,
+ tool_calls=response.tool_calls,
+ )
Review Comment:
It seems we are creating a `ToolRequestEvent` with the id of the
`ChatRequestEvent`? We should never have multiple events with the same id.
##########
python/flink_agents/api/agents/react_agent.py:
##########
@@ -0,0 +1,251 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+#################################################################################
+import json
+from collections.abc import Callable
+from typing import Any, List, Optional, cast
+
+from pydantic import BaseModel
+from pyflink.common import Row
+from pyflink.common.typeinfo import RowTypeInfo
+
+from flink_agents.api.agent import Agent
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
+from flink_agents.api.decorators import action
+from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
+from flink_agents.api.events.event import InputEvent, OutputEvent
+from flink_agents.api.prompts.prompt import Prompt
+from flink_agents.api.resource import ResourceType
+from flink_agents.api.runner_context import RunnerContext
+
+DEFAULT_CHAT_MODEL = "_default_chat_model"
+DEFAULT_SCHEMA_PROMPT = "_default_schema_prompt"
+DEFAULT_USER_PROMPT = "_default_user_prompt"
+OUTPUT_PARSER = "_output_parser"
+OUTPUT_SCHEMA = "_output_schema"
Review Comment:
Should these be private?
##########
python/flink_agents/plan/tests/test_action.py:
##########
@@ -58,14 +60,22 @@ def action() -> Action: # noqa: D103
name="legal",
exec=func,
listen_event_types=[f"{InputEvent.__module__}.{InputEvent.__qualname__}"],
+ params={
Review Comment:
`Action` is not a public api. The question is, do we want to expose the
ability of adding parameters to users. If true, how can user use this feature
via public apis? If not, we should not expose this via runner context, which is
also public api.
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
+ model: str,
+ chat_model: BaseChatModelSetup,
Review Comment:
Why do we need both `model` and `chat_model`?
##########
python/flink_agents/plan/actions/chat_model_action.py:
##########
@@ -15,69 +15,129 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import copy
+from typing import List, cast
+from uuid import UUID
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelSetup
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.event import Event
from flink_agents.api.events.tool_event import ToolRequestEvent,
ToolResponseEvent
+from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.actions.action import Action
from flink_agents.plan.function import PythonFunction
+TOOL_CALL_CONTEXT = "_TOOL_CALL_CONTEXT"
+
+
+def chat(
+ request_id: UUID,
+ model: str,
+ chat_model: BaseChatModelSetup,
+ messages: List[ChatMessage],
+ short_term_memory: MemoryObject,
+ ctx: RunnerContext,
+) -> None:
+ """Chat with llm.
+
+ If there is no tool call generated, we return the chat response event
directly,
+ otherwise, we generate tool request event according to the tool calls in
chat model
+ response, and save the request and response messages in tool call context.
+ """
+ # TODO: support async execution of chat.
+ response = chat_model.chat(messages)
+
+ # generate tool request event according tool calls in response
+ if len(response.tool_calls) > 0:
+ # TODO: Because memory doesn't support remove currently, so we use
+ # dict to store tool context in memory and remove the specific
+ # tool context from dict after consuming. This will cause write and
+ # read amplification for we need get the whole dict and overwrite it
+ # to memory each time we update a specific tool context.
+ # After memory supports remove, we can use
"TOOL_CALL_CONTEXT/request_id"
+ # to store and remove the specific tool context directly.
+
+ # get tool call context
+ tool_call_context = short_term_memory.get(TOOL_CALL_CONTEXT)
+ if not tool_call_context:
+ tool_call_context = {}
+ if request_id not in tool_call_context:
+ tool_call_context[request_id] = copy.deepcopy(messages)
+ # append response to tool call context
+ tool_call_context[request_id].append(response)
+ # update tool call context
+ short_term_memory.set(TOOL_CALL_CONTEXT, tool_call_context)
+ ctx.send_event(
+ ToolRequestEvent(
+ id=request_id,
+ model=model,
+ tool_calls=response.tool_calls,
+ )
+ )
+ # if there is no tool call generated, return chat response directly
+ else:
+ # clear tool call context related to specific request id
+ tool_call_context = short_term_memory.get(TOOL_CALL_CONTEXT)
+ if tool_call_context and request_id in tool_call_context:
+ tool_call_context.pop(request_id)
+ short_term_memory.set(TOOL_CALL_CONTEXT, tool_call_context)
+ ctx.send_event(
+ ChatResponseEvent(
+ request_id=request_id,
+ response=response,
+ )
+ )
+
def process_chat_request_or_tool_response(event: Event, ctx: RunnerContext) ->
None:
- """Built-in action for processing a chat request or tool response."""
+ """Built-in action for processing a chat request or tool response.
+
+ Internally, this action will use short term memory to save the tool call
context,
+ which is a dict mapping request id to chat messages.
+ """
+ short_term_memory = ctx.get_short_term_memory()
if isinstance(event, ChatRequestEvent):
- chat_model = ctx.get_resource(event.model, ResourceType.CHAT_MODEL)
- # TODO: support async execution of chat.
- response = chat_model.chat(event.messages)
- # call tool
- if len(response.tool_calls) > 0:
- for tool_call in response.tool_calls:
- # store the tool call context in short term memory
- state = ctx.get_short_term_memory()
- # TODO: Because memory doesn't support remove currently, so we
use
- # dict to store tool context in memory and remove the specific
- # tool context from dict after consuming. This will cause some
- # overhead for we need get the whole dict and overwrite it to
memory
- # each time we update a specific tool context.
- # After memory supports remove, we can use
- # "__tool_context/tool_call_id" to store and remove the
specific tool
- # context directly.
- if not state.is_exist("__tool_context"):
- state.set("__tool_context", {})
- tool_context = state.get("__tool_context")
- tool_call_id = tool_call["id"]
- tool_context[tool_call_id] = event
- tool_context[tool_call_id].messages.append(response)
- state.set("__tool_context", tool_context)
- ctx.send_event(
- ToolRequestEvent(
- id=tool_call_id,
- tool=tool_call["function"]["name"],
- kwargs=tool_call["function"]["arguments"],
- )
- )
-
- # send response
- else:
- ctx.send_event(ChatResponseEvent(request=event, response=response))
+ chat_model = cast(
+ "BaseChatModelSetup", ctx.get_resource(event.model,
ResourceType.CHAT_MODEL)
+ )
+
+ chat(
+ request_id=event.id,
+ model=event.model,
+ chat_model=chat_model,
+ messages=event.messages,
+ short_term_memory=short_term_memory,
+ ctx=ctx,
+ )
+
elif isinstance(event, ToolResponseEvent):
- state = ctx.get_short_term_memory()
-
- if state.is_exist("__tool_context"):
- tool_context = state.get("__tool_context")
- tool_call_id = event.request.id
- if tool_context is not None and tool_call_id in tool_context:
- # get the specific tool call context from short term memory
- specific_tool_ctx = tool_context.pop(tool_call_id)
- specific_tool_ctx.messages.append(
- ChatMessage(role=MessageRole.TOOL,
content=str(event.response))
- )
- ctx.send_event(specific_tool_ctx)
- # update short term memory to remove the specific tool call
context
- state.set("__tool_context", tool_context)
+ request_id = event.request.id
+ model = event.request.model
+ responses = event.responses
Review Comment:
I assume this is why we need to keep `ToolRequestEvent` in
`ToolResponseEvent`? Alternatively, we can save the tool request id and model
in state, and fetch it when receiving the tool response.
##########
python/flink_agents/api/runner_context.py:
##########
@@ -53,6 +54,16 @@ def get_resource(self, name: str, type: ResourceType) ->
Resource:
The type of the resource.
"""
+ @abstractmethod
+ def get_action_params(self, name: str) -> Dict[str, Any]:
+ """Get additional parameters of action.
+
+ Parameters
+ ----------
+ name : str
+ The name of the action.
+ """
Review Comment:
We should not need the user to pass in the action name. The framework should
know this information.
##########
python/flink_agents/api/runner_context.py:
##########
@@ -53,6 +54,16 @@ def get_resource(self, name: str, type: ResourceType) ->
Resource:
The type of the resource.
"""
+ @abstractmethod
+ def get_action_params(self, name: str) -> Dict[str, Any]:
+ """Get additional parameters of action.
+
+ Parameters
+ ----------
+ name : str
+ The name of the action.
+ """
Review Comment:
And maybe we can have an additional method for getting one specific config /
param, rather than returning all of them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
