Hello. I am trying to write a test case of dynamic execution sequence like the following:
What I want is, the planner_node firstly create the execution sequence, then in execute_node, the LLM parses the single step and create a tool call message, the message is passed to a ToolNode and the tool is invoked. In the real program, there are many tools, I want to dynamically search them and bind them to ToolNode. How to do that ? Anyone has some idea ? Thanks.
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel, Field
from typing import List
from typing_extensions import Annotated
import operator
from model import create_model
@tool
def funca(a:int) -> int:
'''
A simple test tool, raise ValueError if a negative int is received, return `a+1` otherwise.
Args:
a: int parameter.
'''
if a < 0:
raise ValueError("Input value must be non-negative.")
else:
return a+1
@tool
def funcb(content:str) -> str:
'''
A simple test tool, just add `Hello ` ahead of the `content`.
Args:
content: input content.
'''
return "Hello " + content
# Here create a chat model.
model = create_model()
class Sequence(BaseModel):
steps: List[str] = Field(description="The descriptions of execution steps. For example ['call `funca` with argument `10`', 'call `funcb` with argument `'Test'`'] ", default_factory=list)
class MyState(BaseModel):
user_prompt:str = ""
step_i: int = 0
steps: List[str]
results: Annotated[List[str], operator.add]
def planner_node(state: MyState):
system_prompt = ''' You are a planner agent to create the execution sequence.
There are two available execution steps:
- funca(a:int): raise ValueError if `a` is negative, return `a+1` otherwise.
- funcb(content:str): just return `'Hello ' + content`.
You should extracted the sequence of steps from the user's prompt. An example:
User: "run funca with 10, run funcb with 'World', at last run funca with -10."
Returned sequence: ["call funca with argument `10`", "call funcb with argument `'World'`", "call funca with `-10`"]
'''
steps = model.with_structured_output(Sequence).invoke([SystemMessage(content=system_prompt),
HumanMessage(content=state.user_prompt)])
return {
"steps": steps
}
def execute_node(state:MyState):
if state.step_i < len(state.steps):
pass
