mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-20 18:51:03 +02:00
Remove prototypes, no longer needed
This commit is contained in:
Cyber MacGeddon
parent
6d0fc3c89b
commit
a4ace74150
8 changed files with 0 additions and 2154 deletions
|
|
@ -1,302 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
|
||||
class Input:
|
||||
state: dict
|
||||
plan_state: str
|
||||
plan: str
|
||||
iteration: int
|
||||
|
||||
class Output:
|
||||
state: dict
|
||||
thought: str
|
||||
next_state: str
|
||||
plan: str
|
||||
iteration: int
|
||||
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, initial_indent=prefix, subsequent_indent=prefix,
|
||||
width=width
|
||||
)
|
||||
print("\n".join(out))
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
|
||||
class State:
|
||||
pass
|
||||
|
||||
class ActionWikipedia:
|
||||
|
||||
def __init__(self, prompt_client):
|
||||
self.prompt_client = prompt_client
|
||||
|
||||
def act(self, state, topic):
|
||||
|
||||
pr = ibis.Template(
|
||||
"Tell me what you think Wikipedia would say about {{topic}}"
|
||||
)
|
||||
|
||||
input = pr.render({
|
||||
"topic": topic,
|
||||
})
|
||||
|
||||
resp = self.prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": input
|
||||
}
|
||||
)
|
||||
|
||||
state.previous.append(
|
||||
f"Question: {input.strip()}"
|
||||
)
|
||||
|
||||
state.previous.append(
|
||||
f"Answer: {resp.strip()}"
|
||||
)
|
||||
|
||||
return state
|
||||
|
||||
class ActionCalculate:
|
||||
|
||||
def __init__(self, prompt_client):
|
||||
self.prompt_client = prompt_client
|
||||
|
||||
def act(self, state, calc):
|
||||
|
||||
pr = ibis.Template(
|
||||
"Compute this: {{computation}}. Just answer the question without explanation."
|
||||
)
|
||||
|
||||
input = pr.render({
|
||||
"computation": calc,
|
||||
})
|
||||
|
||||
resp = self.prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": input
|
||||
}
|
||||
)
|
||||
|
||||
print(resp)
|
||||
|
||||
state.previous.append(
|
||||
"Calculate: " + calc.strip()
|
||||
)
|
||||
|
||||
state.previous.append(
|
||||
"Answer: " + resp.strip()
|
||||
)
|
||||
|
||||
return state
|
||||
|
||||
class ActionAnswer:
|
||||
|
||||
def __init__(self, prompt_client):
|
||||
self.prompt_client = prompt_client
|
||||
|
||||
def act(self, state, question):
|
||||
|
||||
pr = ibis.Template(
|
||||
"Answer this: {{question}}"
|
||||
)
|
||||
|
||||
input = pr.render({
|
||||
"question": question,
|
||||
})
|
||||
|
||||
resp = self.prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": input
|
||||
}
|
||||
)
|
||||
|
||||
state.facts.append(
|
||||
"Answer to " + input + ":\n" +
|
||||
resp
|
||||
)
|
||||
|
||||
return state
|
||||
|
||||
tools = {
|
||||
"calculate": {
|
||||
"description": "Takes a numeric computation and calculates the answer",
|
||||
"implementation": ActionCalculate,
|
||||
},
|
||||
"cats-knowledge-store": {
|
||||
"description": "Answer questions on Mark's cats using a knowledge store",
|
||||
"implementation": ActionCalculate,
|
||||
},
|
||||
# "wikipedia": {
|
||||
# "description": "Takes a query and looks it up on Wikipedia",
|
||||
# "implementation": ActionWikipedia,
|
||||
# },
|
||||
# "answer": {
|
||||
# "description": "Take a simple question and use the LLM to provide the answer",
|
||||
# "implementation": ActionAnswer,
|
||||
# },
|
||||
# "space-shuttle-rag": {
|
||||
# "description": "Take a question about space shuttles and answer it using the TrustGraph GraphRAG service",
|
||||
# "implementation": ActionAnswer,
|
||||
# }
|
||||
}
|
||||
|
||||
class AgentManager:
|
||||
|
||||
determine_prompt = ibis.Template("""
|
||||
You have access to the following tools:
|
||||
{% for id, tool in tools.items() %}- tool-name: {{id}}
|
||||
tool-description: {{tool.description}}
|
||||
{% endfor %}
|
||||
You operate in a loop. For each iteration of the loop, you take the
|
||||
question and learnt knowledge. If the knowledge is enough to answer
|
||||
the question, you will response with an answer. If the knowledge is not
|
||||
enough, you will response with an action to be invoked to acquire more
|
||||
knowledge.
|
||||
|
||||
Your output must ALWAYS be a well-formed JSON object. Output ONLY a JSON
|
||||
object with no explanatory text or markup formatting. The JSON object
|
||||
can have the following fields:
|
||||
- thought: Mandatory. Your explanation of the current step, including the
|
||||
goal and the reason for choosing this action. A string.
|
||||
- action: An optional field, one of the tool-names above if you are using
|
||||
an action.
|
||||
- action-input: An optional field, input to the selected tool if you are using
|
||||
an action.
|
||||
- failure: a boolean. Set to true if you cannot proceed.
|
||||
- final-answer: Your answer to the question as a string.
|
||||
|
||||
When you know the final answer to the original question, emit a 'final-answer'
|
||||
JSON object with the following fields: thought and final-answer.
|
||||
|
||||
When you don't know the answer but know an action to move towards the
|
||||
solution, emit an 'action' JSON object with the following fields:
|
||||
thought, action, action-input
|
||||
|
||||
When you are stuck and aren't able to move torwards a solution, emit a
|
||||
'failure' JSON object the following fields: thought, failure
|
||||
|
||||
{{previous}}""")
|
||||
|
||||
def __init__(self, tools, prompt_client):
|
||||
|
||||
self.tools = tools
|
||||
self.prompt_client = prompt_client
|
||||
|
||||
def determine(self, state, thought=None):
|
||||
|
||||
input = __class__.determine_prompt.render({
|
||||
"previous": "\n".join(state.previous),
|
||||
"tools": self.tools,
|
||||
})
|
||||
|
||||
print(input)
|
||||
|
||||
resp = self.prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": input
|
||||
}
|
||||
)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
print(resp)
|
||||
# print(resp)
|
||||
resp = json.loads(resp)
|
||||
# print(json.dumps(resp, indent=4))
|
||||
return resp
|
||||
|
||||
def invoke(self, q, thought):
|
||||
|
||||
state = State()
|
||||
state.question = q
|
||||
state.previous = [
|
||||
f"Question: {q}"
|
||||
]
|
||||
|
||||
while True:
|
||||
|
||||
resp = self.determine(state)
|
||||
|
||||
print(resp)
|
||||
|
||||
if "thought" in resp:
|
||||
if thought:
|
||||
thought(resp["thought"])
|
||||
state.previous.append(f"Thought: {resp['thought'].strip()}")
|
||||
|
||||
if "failure" in resp:
|
||||
if resp["failure"]:
|
||||
print("Failed")
|
||||
return "Failed"
|
||||
|
||||
if "final-answer" in resp:
|
||||
return resp["final-answer"]
|
||||
|
||||
if "action" not in resp:
|
||||
raise RuntimeError("Didn't get final-answer/action response")
|
||||
|
||||
print(resp)
|
||||
|
||||
action = resp["action"]
|
||||
|
||||
print(action)
|
||||
|
||||
if action not in self.tools:
|
||||
raise RuntimeError(f"Tool {action} not known")
|
||||
|
||||
if "implementation" not in self.tools[action]:
|
||||
raise RuntimeError(f"Tool {action} not implemented")
|
||||
|
||||
if "action-input" in resp:
|
||||
input = resp["action-input"]
|
||||
else:
|
||||
input = None
|
||||
|
||||
print("Action:", action)
|
||||
print("Action input:", input)
|
||||
print()
|
||||
|
||||
impl_class = self.tools[action]["implementation"]
|
||||
impl = impl_class(self.prompt_client)
|
||||
state = impl.act(state, input)
|
||||
|
||||
time.sleep(2)
|
||||
|
||||
def thought(t):
|
||||
output("\U0001f914... " + t, prefix="? ")
|
||||
print()
|
||||
|
||||
q = """Take the square root of 1600. That number is the number of a US
|
||||
President. Who is that president and how do they relate to the space
|
||||
shuttle programme?
|
||||
"""
|
||||
|
||||
q = "How many eggs do I have if I have 3 eggs and Fred gives me some more eggs. The number of eggs he gives me is the square root of 64."
|
||||
|
||||
q = "Find out the behavioral nature of Mark's cats"
|
||||
|
||||
am = AgentManager(tools, prompt_client)
|
||||
|
||||
output("Q: " + q)
|
||||
print()
|
||||
|
||||
resp = am.invoke(q, thought)
|
||||
|
||||
print(resp)
|
||||
|
||||
|
|
@ -1,184 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
def graph_rag_query(q):
|
||||
resp = rag_client.request(q)
|
||||
return resp
|
||||
|
||||
tools = [
|
||||
{
|
||||
"function": "cats-kb",
|
||||
"description": "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "shuttle-kb",
|
||||
"description": "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "compute",
|
||||
"description": "A computation engine which can answer questions about maths",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The computation"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.function }}",
|
||||
"description": "{{ tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately and only - with a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"action-input": the input to the action,
|
||||
"observation": "the result of the action",
|
||||
},
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/action-input or the final-answer. Pause after providing each action.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input: {% for h in history %}{{h}}
|
||||
{% endfor %}"""
|
||||
|
||||
tpl = ibis.Template(template)
|
||||
|
||||
history = []
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
output(wrap(q))
|
||||
print()
|
||||
|
||||
for iter in range(0, 10):
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": q,
|
||||
"tool_names": ",".join([t["function"] for t in tools]),
|
||||
"history": json.dumps(history, indent=2),
|
||||
})
|
||||
|
||||
print(prompt)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
print(resp)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
obj = json.loads(resp)
|
||||
|
||||
output(obj["thought"], "? ")
|
||||
print()
|
||||
|
||||
if "final-answer" in obj:
|
||||
history.append(obj)
|
||||
break
|
||||
|
||||
if "action" in obj:
|
||||
if obj["action"] == "cats-kb":
|
||||
ans = graph_rag_query(obj["action-input"])
|
||||
obj["observation"] = ans.strip()
|
||||
elif obj["action"] == "compute":
|
||||
if "196" in obj["action-input"]:
|
||||
obj["observation"] = "196 > 1.319501555"
|
||||
else:
|
||||
obj["observation"] = "The answer is 1.319501155"
|
||||
elif obj["action"] == "shuttle-kb":
|
||||
ans = graph_rag_query(obj["action-input"])
|
||||
obj["observation"] = ans.strip()
|
||||
else:
|
||||
raise RuntimeError("Unknown action:", obj["action"])
|
||||
|
||||
output(obj["observation"], "! ")
|
||||
print()
|
||||
|
||||
history.append(obj)
|
||||
continue
|
||||
|
||||
raise RuntimeError("Iteration output is not action or final-answer")
|
||||
|
||||
output(obj["final-answer"], "< ")
|
||||
|
||||
|
|
@ -1,198 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
def graph_rag_query(q):
|
||||
resp = rag_client.request(q)
|
||||
return resp
|
||||
|
||||
tools = [
|
||||
{
|
||||
"function": "cats-kb",
|
||||
"description": "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "shuttle-kb",
|
||||
"description": "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "compute",
|
||||
"description": "A computation engine which can answer questions about maths",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The computation"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.function }}",
|
||||
"description": "{{ tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately and only - with a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or the final-answer. Pause after providing each action.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
{% for h in history %}{{h}}
|
||||
|
||||
{% endfor %}"""
|
||||
|
||||
tpl = ibis.Template(template)
|
||||
|
||||
history = []
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
output(wrap(q))
|
||||
print()
|
||||
|
||||
for iter in range(0, 10):
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": q,
|
||||
"tool_names": ",".join([t["function"] for t in tools]),
|
||||
"history": [json.dumps(h, indent=2) for h in history],
|
||||
})
|
||||
|
||||
print("-" * 76)
|
||||
print(prompt)
|
||||
print("-" * 76)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
print("-" * 76)
|
||||
print(resp)
|
||||
print("-" * 76)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
obj = json.loads(resp)
|
||||
|
||||
output(obj["thought"], "? ")
|
||||
print()
|
||||
|
||||
if "final-answer" in obj:
|
||||
history.append(obj)
|
||||
break
|
||||
|
||||
if "action" in obj:
|
||||
if obj["action"] == "cats-kb":
|
||||
ans = graph_rag_query(obj["arguments"]["query"])
|
||||
obj["observation"] = ans.strip()
|
||||
elif obj["action"] == "compute":
|
||||
if "196" in obj["arguments"]["query"]:
|
||||
obj["observation"] = "196 > 1.319501555"
|
||||
else:
|
||||
obj["observation"] = "The answer is 1.319501155"
|
||||
elif obj["action"] == "shuttle-kb":
|
||||
ans = graph_rag_query(obj["arguments"]["query"])
|
||||
obj["observation"] = ans.strip()
|
||||
else:
|
||||
raise RuntimeError("Unknown action:", obj["action"])
|
||||
|
||||
output(obj["observation"], "! ")
|
||||
print()
|
||||
|
||||
history.append(obj)
|
||||
|
||||
print("-" * 76)
|
||||
print(history)
|
||||
print("-" * 76)
|
||||
|
||||
continue
|
||||
|
||||
raise RuntimeError("Iteration output is not action or final-answer")
|
||||
|
||||
output(obj["final-answer"], "< ")
|
||||
|
||||
|
|
@ -1,197 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
def graph_rag_query(q):
|
||||
resp = rag_client.request(q)
|
||||
return resp
|
||||
|
||||
tools = [
|
||||
{
|
||||
"function": "cats-kb",
|
||||
"description": "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "shuttle-kb",
|
||||
"description": "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The search query string"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "compute",
|
||||
"description": "A computation engine which can answer questions about maths",
|
||||
"arguments": [
|
||||
{
|
||||
"name": "query",
|
||||
"type": "string",
|
||||
"description": "The computation"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.function }}",
|
||||
"description": "{{ tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately and only - with a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or the final-answer. Pause after providing each action.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
{% for h in history %}{{h}}
|
||||
|
||||
{% endfor %}"""
|
||||
|
||||
tpl = ibis.Template(template)
|
||||
|
||||
history = []
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
output(wrap(q))
|
||||
print()
|
||||
|
||||
for iter in range(0, 10):
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": q,
|
||||
"tool_names": ",".join([t["function"] for t in tools]),
|
||||
"history": [json.dumps(h, indent=2) for h in history],
|
||||
})
|
||||
|
||||
print("-" * 76)
|
||||
print(prompt)
|
||||
print("-" * 76)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
print("-" * 76)
|
||||
print(resp)
|
||||
print("-" * 76)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
obj = json.loads(resp)
|
||||
|
||||
output(obj["thought"], "? ")
|
||||
print()
|
||||
|
||||
if "final-answer" in obj:
|
||||
history.append(obj)
|
||||
break
|
||||
|
||||
if "action" in obj:
|
||||
if obj["action"] == "cats-kb":
|
||||
ans = graph_rag_query(obj["arguments"]["query"])
|
||||
obj["observation"] = ans.strip()
|
||||
elif obj["action"] == "compute":
|
||||
if "196" in obj["arguments"]["query"]:
|
||||
obj["observation"] = "196 > 1.319501555"
|
||||
else:
|
||||
obj["observation"] = "The answer is 1.319501155"
|
||||
elif obj["action"] == "shuttle-kb":
|
||||
ans = graph_rag_query(obj["arguments"]["query"])
|
||||
obj["observation"] = ans.strip()
|
||||
else:
|
||||
raise RuntimeError("Unknown action:", obj["action"])
|
||||
|
||||
output(obj["observation"], "! ")
|
||||
print()
|
||||
|
||||
history.append(obj)
|
||||
|
||||
print("-" * 76)
|
||||
print(history)
|
||||
print("-" * 76)
|
||||
|
||||
continue
|
||||
|
||||
raise RuntimeError("Iteration output is not action or final-answer")
|
||||
|
||||
output(obj["final-answer"], "< ")
|
||||
|
||||
318
agent/plan-b-1
318
agent/plan-b-1
|
|
@ -1,318 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
import sys
|
||||
import base64
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
def graph_rag_query(q):
|
||||
resp = rag_client.request(q)
|
||||
return resp
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Argument:
|
||||
name : str
|
||||
type : str
|
||||
description : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Tool:
|
||||
name : str
|
||||
description : str
|
||||
arguments : list[Argument]
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Action:
|
||||
thought : str
|
||||
name : str
|
||||
arguments : dict
|
||||
observation : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Final:
|
||||
thought : str
|
||||
final : str
|
||||
|
||||
class CatsKb:
|
||||
tool = Tool(
|
||||
name = "cats-kb",
|
||||
description = "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class ShuttleKb:
|
||||
tool = Tool(
|
||||
name = "shuttle-kb",
|
||||
description = "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class Compute:
|
||||
tool = Tool(
|
||||
name = "compute",
|
||||
description = "A computation engine which can answer questions about maths and computation",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "computation",
|
||||
type = "string",
|
||||
description = "The computation to solve"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(
|
||||
"question", { "question": arguments.get("computation") }
|
||||
)
|
||||
|
||||
# t = CatsKb(rag_client)
|
||||
# print(t.invoke(query="How many cats does Mark have?"))
|
||||
|
||||
# t = Compute(prompt_client)
|
||||
# print(t.invoke(computation="12 + 4"))
|
||||
|
||||
tools = [
|
||||
CatsKb(rag_client),
|
||||
ShuttleKb(rag_client),
|
||||
Compute(prompt_client),
|
||||
]
|
||||
|
||||
class AgentManager:
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.tool.name }}",
|
||||
"description": "{{ tool.tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately and only - with a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or the final-answer. Pause after providing each action.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
{% for h in history %}
|
||||
{
|
||||
"action": "{{h.action}}",
|
||||
"arguments": [
|
||||
{% for k, v in h.arguments.items() %} {
|
||||
"{{k}}": "{{v}}",
|
||||
{%endfor%} }
|
||||
],
|
||||
"observation": "{{h.observation}}"
|
||||
}
|
||||
{% endfor %}"""
|
||||
|
||||
def __init__(self, client, tools, question):
|
||||
self.client = client
|
||||
self.tools = tools
|
||||
self.history = []
|
||||
self.question = question
|
||||
|
||||
def iterate(self):
|
||||
|
||||
tpl = ibis.Template(self.template)
|
||||
|
||||
tools = self.tools
|
||||
|
||||
tool_names = ",".join([
|
||||
t.tool.name for t in self.tools
|
||||
])
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": self.question,
|
||||
"tool_names": tool_names,
|
||||
"history": [
|
||||
{
|
||||
"thought": h.thought,
|
||||
"action": h.name,
|
||||
"arguments": h.arguments,
|
||||
"observation": h.observation,
|
||||
}
|
||||
for h in self.history
|
||||
],
|
||||
})
|
||||
|
||||
print(prompt)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
print("---")
|
||||
print(resp)
|
||||
print("---")
|
||||
|
||||
print(base64.b64encode(resp.encode("utf-8")).decode("utf-8"))
|
||||
obj = json.loads(resp)
|
||||
|
||||
if obj.get("final-answer"):
|
||||
|
||||
a = Final(
|
||||
thought = obj.get("thought"),
|
||||
final = obj.get("final-answer"),
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
else:
|
||||
|
||||
a = Action(
|
||||
thought = obj.get("thought"),
|
||||
name = obj.get("action"),
|
||||
arguments = obj.get("arguments"),
|
||||
observation = "FIXME"
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
def think(self, thought):
|
||||
|
||||
print(thought)
|
||||
|
||||
output(thought, "? ")
|
||||
print()
|
||||
|
||||
def invoke(self):
|
||||
|
||||
for i in range(0, 10):
|
||||
|
||||
print("-----------------------")
|
||||
print("Iter", i)
|
||||
|
||||
act = self.iterate()
|
||||
print(act)
|
||||
|
||||
if isinstance(act, Final):
|
||||
|
||||
self.think(act.thought)
|
||||
return act.final
|
||||
|
||||
else:
|
||||
|
||||
self.think(act.thought)
|
||||
|
||||
print(act.name)
|
||||
|
||||
action = None
|
||||
|
||||
for tool in self.tools:
|
||||
if tool.tool.name == act.name:
|
||||
action = tool
|
||||
|
||||
if action is None:
|
||||
raise RuntimeError(f"No action for {act.name}!")
|
||||
|
||||
resp = action.invoke(**act.arguments)
|
||||
|
||||
resp = resp.strip()
|
||||
|
||||
print(resp)
|
||||
|
||||
act.observation = resp
|
||||
|
||||
print("act>", act)
|
||||
|
||||
self.history.append(act)
|
||||
|
||||
print(self.history)
|
||||
|
||||
raise RuntimeError("Too many iterations")
|
||||
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
am = AgentManager(prompt_client, tools, q)
|
||||
|
||||
act = am.invoke()
|
||||
print(act)
|
||||
|
||||
312
agent/plan-b-2
312
agent/plan-b-2
|
|
@ -1,312 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
import sys
|
||||
import base64
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
def graph_rag_query(q):
|
||||
resp = rag_client.request(q)
|
||||
return resp
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Argument:
|
||||
name : str
|
||||
type : str
|
||||
description : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Tool:
|
||||
name : str
|
||||
description : str
|
||||
arguments : list[Argument]
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Action:
|
||||
thought : str
|
||||
name : str
|
||||
arguments : dict
|
||||
observation : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Final:
|
||||
thought : str
|
||||
final : str
|
||||
|
||||
class CatsKb:
|
||||
tool = Tool(
|
||||
name = "cats-kb",
|
||||
description = "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class ShuttleKb:
|
||||
tool = Tool(
|
||||
name = "shuttle-kb",
|
||||
description = "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class Compute:
|
||||
tool = Tool(
|
||||
name = "compute",
|
||||
description = "A computation engine which can answer questions about maths and computation",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "computation",
|
||||
type = "string",
|
||||
description = "The computation to solve"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(
|
||||
"question", { "question": arguments.get("computation") }
|
||||
)
|
||||
|
||||
tools = [
|
||||
CatsKb(rag_client),
|
||||
ShuttleKb(rag_client),
|
||||
Compute(prompt_client),
|
||||
]
|
||||
|
||||
class AgentManager:
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.tool.name }}",
|
||||
"description": "{{ tool.tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately with exactly one action - a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or the final-answer. Pause after providing one action or final answer.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
{% for h in history %}
|
||||
{
|
||||
"action": "{{h.action}}",
|
||||
"arguments": [
|
||||
{% for k, v in h.arguments.items() %} {
|
||||
"{{k}}": "{{v}}",
|
||||
{%endfor%} }
|
||||
],
|
||||
"observation": "{{h.observation}}"
|
||||
}
|
||||
{% endfor %}"""
|
||||
|
||||
def __init__(self, client, tools, question):
|
||||
self.client = client
|
||||
self.tools = tools
|
||||
self.history = []
|
||||
self.question = question
|
||||
|
||||
def iterate(self):
|
||||
|
||||
tpl = ibis.Template(self.template)
|
||||
|
||||
tools = self.tools
|
||||
|
||||
tool_names = ",".join([
|
||||
t.tool.name for t in self.tools
|
||||
])
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": self.question,
|
||||
"tool_names": tool_names,
|
||||
"history": [
|
||||
{
|
||||
"thought": h.thought,
|
||||
"action": h.name,
|
||||
"arguments": h.arguments,
|
||||
"observation": h.observation,
|
||||
}
|
||||
for h in self.history
|
||||
],
|
||||
})
|
||||
|
||||
print(prompt)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
print("---")
|
||||
print(resp)
|
||||
print("---")
|
||||
|
||||
print(base64.b64encode(resp.encode("utf-8")).decode("utf-8"))
|
||||
obj = json.loads(resp)
|
||||
|
||||
if obj.get("final-answer"):
|
||||
|
||||
a = Final(
|
||||
thought = obj.get("thought"),
|
||||
final = obj.get("final-answer"),
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
else:
|
||||
|
||||
a = Action(
|
||||
thought = obj.get("thought"),
|
||||
name = obj.get("action"),
|
||||
arguments = obj.get("arguments"),
|
||||
observation = "FIXME"
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
def think(self, thought):
|
||||
|
||||
print(thought)
|
||||
|
||||
output(thought, "? ")
|
||||
print()
|
||||
|
||||
def invoke(self):
|
||||
|
||||
for i in range(0, 10):
|
||||
|
||||
print("-----------------------")
|
||||
print("Iter", i)
|
||||
|
||||
act = self.iterate()
|
||||
print(act)
|
||||
|
||||
if isinstance(act, Final):
|
||||
|
||||
self.think(act.thought)
|
||||
return act.final
|
||||
|
||||
else:
|
||||
|
||||
self.think(act.thought)
|
||||
|
||||
print(act.name)
|
||||
|
||||
action = None
|
||||
|
||||
for tool in self.tools:
|
||||
if tool.tool.name == act.name:
|
||||
action = tool
|
||||
|
||||
if action is None:
|
||||
raise RuntimeError(f"No action for {act.name}!")
|
||||
|
||||
resp = action.invoke(**act.arguments)
|
||||
|
||||
resp = resp.strip()
|
||||
|
||||
print(resp)
|
||||
|
||||
act.observation = resp
|
||||
|
||||
print("act>", act)
|
||||
|
||||
self.history.append(act)
|
||||
|
||||
print(self.history)
|
||||
|
||||
raise RuntimeError("Too many iterations")
|
||||
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
am = AgentManager(prompt_client, tools, q)
|
||||
|
||||
act = am.invoke()
|
||||
print(act)
|
||||
|
||||
311
agent/plan-b-3
311
agent/plan-b-3
|
|
@ -1,311 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
import sys
|
||||
import base64
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Argument:
|
||||
name : str
|
||||
type : str
|
||||
description : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Tool:
|
||||
name : str
|
||||
description : str
|
||||
arguments : list[Argument]
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Action:
|
||||
thought : str
|
||||
name : str
|
||||
arguments : dict
|
||||
observation : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Final:
|
||||
thought : str
|
||||
final : str
|
||||
|
||||
class CatsKb:
|
||||
tool = Tool(
|
||||
name = "cats-kb",
|
||||
description = "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class ShuttleKb:
|
||||
tool = Tool(
|
||||
name = "shuttle-kb",
|
||||
description = "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class Compute:
|
||||
tool = Tool(
|
||||
name = "compute",
|
||||
description = "A computation engine which can answer questions about maths and computation",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "computation",
|
||||
type = "string",
|
||||
description = "The computation to solve"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(
|
||||
"question", { "question": arguments.get("computation") }
|
||||
)
|
||||
|
||||
tools = [
|
||||
CatsKb(rag_client),
|
||||
ShuttleKb(rag_client),
|
||||
Compute(prompt_client),
|
||||
]
|
||||
|
||||
class AgentManager:
|
||||
|
||||
template="""Answer the following questions as best you can. You have access to the following tools:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.tool.name }}",
|
||||
"description": "{{ tool.tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
To call a function, respond - immediately and only - with a JSON object of the following format:
|
||||
{
|
||||
"action": "function_name",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
Each step has the following format in your output:
|
||||
|
||||
{
|
||||
"thought": "you should always think about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
... (this JSON object can repeat N times)
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or the final-answer. Pause after providing each action.
|
||||
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
{% for h in history %}
|
||||
{
|
||||
"action": "{{h.action}}",
|
||||
"arguments": [
|
||||
{% for k, v in h.arguments.items() %} {
|
||||
"{{k}}": "{{v}}",
|
||||
{%endfor%} }
|
||||
],
|
||||
"observation": "{{h.observation}}"
|
||||
}
|
||||
{% endfor %}"""
|
||||
|
||||
def __init__(self, client, tools, question):
|
||||
self.client = client
|
||||
self.tools = tools
|
||||
self.history = []
|
||||
self.question = question
|
||||
|
||||
def iterate(self):
|
||||
|
||||
tpl = ibis.Template(self.template)
|
||||
|
||||
tools = self.tools
|
||||
|
||||
tool_names = ",".join([
|
||||
t.tool.name for t in self.tools
|
||||
])
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": self.question,
|
||||
"tool_names": tool_names,
|
||||
"history": [
|
||||
{
|
||||
"thought": h.thought,
|
||||
"action": h.name,
|
||||
"arguments": h.arguments,
|
||||
"observation": h.observation,
|
||||
}
|
||||
for h in self.history
|
||||
],
|
||||
})
|
||||
|
||||
# print(prompt)
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
# print("---")
|
||||
# print(resp)
|
||||
# print("---")
|
||||
|
||||
# print(base64.b64encode(resp.encode("utf-8")).decode("utf-8"))
|
||||
obj = json.loads(resp)
|
||||
|
||||
if obj.get("final-answer"):
|
||||
|
||||
a = Final(
|
||||
thought = obj.get("thought"),
|
||||
final = obj.get("final-answer"),
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
else:
|
||||
|
||||
a = Action(
|
||||
thought = obj.get("thought"),
|
||||
name = obj.get("action"),
|
||||
arguments = obj.get("arguments"),
|
||||
observation = "FIXME"
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
def think(self, thought):
|
||||
|
||||
# print(thought)
|
||||
|
||||
output(wrap(thought), "? ")
|
||||
print()
|
||||
|
||||
def invoke(self):
|
||||
|
||||
for i in range(0, 10):
|
||||
|
||||
# print("-----------------------")
|
||||
# print("Iter", i)
|
||||
|
||||
act = self.iterate()
|
||||
# print(act)
|
||||
|
||||
if isinstance(act, Final):
|
||||
|
||||
self.think(act.thought)
|
||||
return act.final
|
||||
|
||||
else:
|
||||
|
||||
self.think(act.thought)
|
||||
|
||||
# print(act.name)
|
||||
|
||||
action = None
|
||||
|
||||
for tool in self.tools:
|
||||
if tool.tool.name == act.name:
|
||||
action = tool
|
||||
|
||||
if action is None:
|
||||
raise RuntimeError(f"No action for {act.name}!")
|
||||
|
||||
resp = action.invoke(**act.arguments)
|
||||
|
||||
resp = resp.strip()
|
||||
|
||||
# print(resp)
|
||||
|
||||
act.observation = resp
|
||||
|
||||
# print("act>", act)
|
||||
|
||||
self.history.append(act)
|
||||
|
||||
# print(self.history)
|
||||
|
||||
raise RuntimeError("Too many iterations")
|
||||
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number higher than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
output(wrap(q))
|
||||
print()
|
||||
|
||||
am = AgentManager(prompt_client, tools, q)
|
||||
|
||||
act = am.invoke()
|
||||
print(act)
|
||||
|
||||
332
agent/plan-b-4
332
agent/plan-b-4
|
|
@ -1,332 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from trustgraph.clients.prompt_client import PromptClient
|
||||
from trustgraph.clients.graph_rag_client import GraphRagClient
|
||||
import json
|
||||
import textwrap
|
||||
import ibis
|
||||
import textwrap
|
||||
import time
|
||||
import dataclasses
|
||||
import base64
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
#logging.basicConfig(level=logging.INFO)
|
||||
|
||||
def wrap(text, width=75):
|
||||
|
||||
if text is None: text = "n/a"
|
||||
|
||||
out = textwrap.wrap(
|
||||
text, width=width
|
||||
)
|
||||
return "\n".join(out)
|
||||
|
||||
def output(text, prefix="> ", width=78):
|
||||
|
||||
out = textwrap.indent(
|
||||
text, prefix=prefix
|
||||
)
|
||||
print(out)
|
||||
|
||||
pulsar_host = "pulsar://localhost:6650"
|
||||
|
||||
prompt_client = PromptClient(pulsar_host=pulsar_host)
|
||||
rag_client = GraphRagClient(pulsar_host=pulsar_host)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Argument:
|
||||
name : str
|
||||
type : str
|
||||
description : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Tool:
|
||||
name : str
|
||||
description : str
|
||||
arguments : list[Argument]
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Action:
|
||||
thought : str
|
||||
name : str
|
||||
arguments : dict
|
||||
observation : str
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Final:
|
||||
thought : str
|
||||
final : str
|
||||
|
||||
class CatsKb:
|
||||
tool = Tool(
|
||||
name = "cats-kb",
|
||||
description = "Query a knowledge base with information about Mark's cats. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class ShuttleKb:
|
||||
tool = Tool(
|
||||
name = "shuttle-kb",
|
||||
description = "Query a knowledge base with information about the space shuttle. The query should be a simple natural language question",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "query",
|
||||
type = "string",
|
||||
description = "The search query string"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(arguments.get("query"))
|
||||
|
||||
class Compute:
|
||||
tool = Tool(
|
||||
name = "compute",
|
||||
description = "A computation engine which can answer questions about maths and computation",
|
||||
arguments = [
|
||||
Argument(
|
||||
name = "computation",
|
||||
type = "string",
|
||||
description = "The computation to solve"
|
||||
)
|
||||
]
|
||||
)
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
def invoke(self, **arguments):
|
||||
return self.client.request(
|
||||
"question", { "question": arguments.get("computation") }
|
||||
)
|
||||
|
||||
tools = [
|
||||
CatsKb(rag_client),
|
||||
ShuttleKb(rag_client),
|
||||
Compute(prompt_client),
|
||||
]
|
||||
|
||||
class AgentManager:
|
||||
|
||||
template="""Answer the following questions as best you can. You have
|
||||
access to the following functions:
|
||||
|
||||
{% for tool in tools %}{
|
||||
"function": "{{ tool.tool.name }}",
|
||||
"description": "{{ tool.tool.description }}",
|
||||
"arguments": [
|
||||
{% for arg in tool.tool.arguments %} {
|
||||
"name": "{{ arg.name }}",
|
||||
"type": "{{ arg.type }}",
|
||||
"description": "{{ arg.description }}",
|
||||
}
|
||||
{% endfor %}
|
||||
]
|
||||
}
|
||||
{% endfor %}
|
||||
|
||||
You can either choose to call a function to get more information, or
|
||||
return a final answer.
|
||||
|
||||
To call a function, respond with a JSON object of the following format:
|
||||
|
||||
{
|
||||
"thought": "your thought about what to do",
|
||||
"action": "the action to take, should be one of [{{tool_names}}]",
|
||||
"arguments": {
|
||||
"argument1": "argument_value",
|
||||
"argument2": "argument_value"
|
||||
}
|
||||
}
|
||||
|
||||
To provide a final answer, response a JSON object of the following format:
|
||||
|
||||
{
|
||||
"thought": "I now know the final answer",
|
||||
"final-answer": "the final answer to the original input question"
|
||||
}
|
||||
|
||||
Previous steps are included in the input. Each step has the following
|
||||
format in your output:
|
||||
|
||||
{
|
||||
"thought": "your thought about what to do",
|
||||
"action": "the action taken",
|
||||
"arguments": {
|
||||
"argument1": action argument,
|
||||
"argument2": action argument2
|
||||
},
|
||||
"observation": "the result of the action",
|
||||
}
|
||||
|
||||
Respond by describing either one single thought/action/arguments or
|
||||
the final-answer. Pause after providing one action or final-answer.
|
||||
Begin!
|
||||
|
||||
Question: {{question}}
|
||||
|
||||
Input:
|
||||
|
||||
{% for h in history %}
|
||||
{
|
||||
"action": "{{h.action}}",
|
||||
"arguments": [
|
||||
{% for k, v in h.arguments.items() %} {
|
||||
"{{k}}": "{{v}}",
|
||||
{%endfor%} }
|
||||
],
|
||||
"observation": "{{h.observation}}"
|
||||
}
|
||||
{% endfor %}"""
|
||||
|
||||
def __init__(self, client, tools, question):
|
||||
self.client = client
|
||||
self.tools = tools
|
||||
self.history = []
|
||||
self.question = question
|
||||
|
||||
def iterate(self):
|
||||
|
||||
tpl = ibis.Template(self.template)
|
||||
|
||||
tools = self.tools
|
||||
|
||||
tool_names = ",".join([
|
||||
t.tool.name for t in self.tools
|
||||
])
|
||||
|
||||
prompt = tpl.render({
|
||||
"tools": tools,
|
||||
"question": self.question,
|
||||
"tool_names": tool_names,
|
||||
"history": [
|
||||
{
|
||||
"thought": h.thought,
|
||||
"action": h.name,
|
||||
"arguments": h.arguments,
|
||||
"observation": h.observation,
|
||||
}
|
||||
for h in self.history
|
||||
],
|
||||
})
|
||||
|
||||
logger.info(f"prompt: {prompt}")
|
||||
|
||||
resp = prompt_client.request(
|
||||
"question",
|
||||
{
|
||||
"question": prompt
|
||||
}
|
||||
)
|
||||
|
||||
resp = resp.replace("```json", "")
|
||||
resp = resp.replace("```", "")
|
||||
|
||||
logger.info(f"response: {resp}")
|
||||
|
||||
obj = json.loads(resp)
|
||||
|
||||
if obj.get("final-answer"):
|
||||
|
||||
a = Final(
|
||||
thought = obj.get("thought"),
|
||||
final = obj.get("final-answer"),
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
else:
|
||||
|
||||
a = Action(
|
||||
thought = obj.get("thought"),
|
||||
name = obj.get("action"),
|
||||
arguments = obj.get("arguments"),
|
||||
observation = "FIXME"
|
||||
)
|
||||
|
||||
return a
|
||||
|
||||
def think(self, thought):
|
||||
|
||||
logger.info(f"thought: {thought}")
|
||||
|
||||
output(wrap(thought), "? ")
|
||||
print()
|
||||
|
||||
def observe(self, observation):
|
||||
|
||||
logger.info(f"observe: {observation}")
|
||||
|
||||
output(wrap(observation), "! ")
|
||||
print()
|
||||
|
||||
def invoke(self):
|
||||
|
||||
for i in range(0, 10):
|
||||
|
||||
act = self.iterate()
|
||||
logger.info(f"act: {act}")
|
||||
|
||||
if isinstance(act, Final):
|
||||
|
||||
self.think(act.thought)
|
||||
return act.final
|
||||
|
||||
else:
|
||||
|
||||
self.think(act.thought)
|
||||
|
||||
action = None
|
||||
|
||||
for tool in self.tools:
|
||||
if tool.tool.name == act.name:
|
||||
action = tool
|
||||
|
||||
if action is None:
|
||||
raise RuntimeError(f"No action for {act.name}!")
|
||||
|
||||
resp = action.invoke(**act.arguments)
|
||||
|
||||
resp = resp.strip()
|
||||
|
||||
logger.info(f"resp: {resp}")
|
||||
|
||||
self.observe(resp)
|
||||
|
||||
act.observation = resp
|
||||
|
||||
logger.info(f"iter: {act}")
|
||||
|
||||
self.history.append(act)
|
||||
|
||||
raise RuntimeError("Too many iterations")
|
||||
|
||||
|
||||
q = "How many cats does Mark have? Calculate that number raised to 0.4 power. Is that number lower than the numeric part of the mission identifier of the Space Shuttle Challenger on its last mission? If so, give me an apple pie recipe, otherwise return a poem about cheese."
|
||||
|
||||
logger.info(f"q: {q}")
|
||||
|
||||
output(wrap(q))
|
||||
print()
|
||||
|
||||
am = AgentManager(prompt_client, tools, q)
|
||||
|
||||
resp = am.invoke()
|
||||
|
||||
logger.info(f"answer: {resp}")
|
||||
|
||||
output(resp, "< ")
|
||||
print()
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue