apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-21 19:21:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Now working with the tool descriptor arguments

Browse source
This commit is contained in:
Cyber MacGeddon 2024-11-10 10:52:28 +00:00
parent 3f60ba540b
commit 7a52e94779
2 changed files with 34 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

36
trustgraph-flow/trustgraph/agent/react/agent_manager.py
View file

@ -13,10 +13,10 @@ class AgentManager:
access to the following functions:
{% for tool in tools %}{
"function": "{{ tool.tool.name }}",
"description": "{{ tool.tool.description }}",
"function": "{{ tool.name }}",
"description": "{{ tool.description }}",
"arguments": [
{% for arg in tool.tool.arguments %} {
{% for arg in tool.arguments %} {
"name": "{{ arg.name }}",
"type": "{{ arg.type }}",
"description": "{{ arg.description }}",
@ -91,11 +91,25 @@ Input:
tools = self.tools
tool_names = ",".join([
t.tool.name for t in self.tools
t for t in self.tools.keys()
])
prompt = tpl.render({
"tools": tools,
"tools": [
{
"name": tool.name,
"description": tool.description,
"arguments": [
{
"name": arg.name,
"type": arg.type,
"description": arg.description
}
for arg in tool.arguments.values()
]
}
for tool in self.tools.values()
],
"question": question,
"tool_names": tool_names,
"history": [
@ -159,16 +173,12 @@ Input:
think(act.thought)
action = None
for tool in self.tools:
if tool.tool.name == act.name:
action = tool
if action is None:
if act.name in self.tools:
action = self.tools[act.name]
else:
raise RuntimeError(f"No action for {act.name}!")
resp = action.invoke(**act.arguments)
resp = action.implementation.invoke(**act.arguments)
resp = resp.strip()

27
trustgraph-flow/trustgraph/agent/react/service.py
View file

@ -24,7 +24,8 @@ from ... clients.graph_rag_client import GraphRagClient
from . tools import KnowledgeQueryImpl, TextCompletionImpl
from . agent_manager import AgentManager
from . types import Final, Action, Tool
from . types import Final, Action, Tool, Argument
module = ".".join(__name__.split(".")[1:-1])
@ -55,9 +56,9 @@ class Processor(ConsumerProducer):
)
if ttoks[0] == "knowledge-query":
impl = KnowledgeQueryImpl
impl = KnowledgeQueryImpl(self)
elif ttoks[0] == "text-completion":
impl = TextCompletionImpl
impl = TextCompletionImpl(self)
else:
raise RuntimeError(
f"Tool-kind {ttoks[0]} not known"
@ -111,15 +112,13 @@ class Processor(ConsumerProducer):
)
if toks[0] not in tool_base:
raise RuntimeError(f"Description, tool {toks[0]} not known")
tool_base[toks[0]].arguments[ttoks[0]] = {
"name": ttoks[0],
"type": ttoks[1],
"description": ttoks[2],
}
tool_base[toks[0]].arguments[ttoks[0]] = Argument(
name = ttoks[0],
type = ttoks[1],
description = ttoks[2]
)
print(json.dumps({k: str(v) for k, v in tool_base.items()}, indent=4))
sys.exit(0)
# print(json.dumps({k: str(v) for k, v in tool_base.items()}, indent=4))
input_queue = params.get("input_queue", default_input_queue)
output_queue = params.get("output_queue", default_output_queue)
@ -186,11 +185,7 @@ class Processor(ConsumerProducer):
schema=JsonSchema(AgentRequest),
)
tools = [
# CatsKb(self), ShuttleKb(self), Compute(self),
]
self.agent = AgentManager(self, tools)
self.agent = AgentManager(self, tool_base)
def parse_json(self, text):
json_match = re.search(r'```(?:json)?(.*?)```', text, re.DOTALL)