-
Notifications
You must be signed in to change notification settings - Fork 1
/
code-generator-agent.py
110 lines (75 loc) · 2.76 KB
/
code-generator-agent.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import openai
from toolhouse import Toolhouse
from uagents import Agent, Context, Model, Protocol
from uagents.models import ErrorMessage
from uagents.setup import fund_agent_if_low
from ai_engine import UAgentResponse, UAgentResponseType
# Set the OpenAI API key correctly
openai.api_key = '<YOUR_OPENAI_API_KEY>'
th = Toolhouse(access_token='<YOUR_TOOLHOUSEAI_API_KEY>', provider="openai")
AGENT_MAILBOX_KEY = "911d1aac-059c-4cae-ab16-cfc75362953f"
class ToolHouseAIRequest(Model):
query: str
agent = Agent(
name="toolhouseai-test-agent",
seed="toolhouseai-seed",
mailbox=f"{AGENT_MAILBOX_KEY}@https://agentverse.ai"
)
fund_agent_if_low(agent.wallet.address())
toolhouseai_proto = Protocol(name="ToolhouseAI-Protocol", version="0.1.0")
async def get_answer(query):
# Define the OpenAI model we want to use
MODEL = 'gpt-4o-mini'
messages = [{
"role": "user",
"content": query
}]
response = openai.chat.completions.create(
model=MODEL,
messages=messages,
# Passes Code Execution as a tool
tools=th.get_tools()
)
# Print the generated code first
generated_code = response.choices[0].message.content
print("Generated Code:\n", generated_code)
# Runs the Code Execution tool, gets the result,
# and appends it to the context
messages += th.run_tools(response)
final_response = openai.chat.completions.create(
model=MODEL,
messages=messages,
tools=th.get_tools()
)
# Prints the execution result
print("Execution Result:\n", final_response.choices[0].message.content)
execution_result = final_response.choices[0].message.content
if "```python" in execution_result and "```" in execution_result:
start = execution_result.find("```python") + len("```python")
end = execution_result.find("```", start)
execution_result_code = execution_result[start:end].strip()
return execution_result_code
@agent.on_event("startup")
async def introduce(ctx: Context):
ctx.logger.info(ctx.agent.address)
@toolhouseai_proto.on_message(ToolHouseAIRequest, replies={UAgentResponse, ErrorMessage})
async def handle_request(ctx: Context, sender: str, msg: ToolHouseAIRequest):
ctx.logger.info(f"Received query : {msg.query}")
try:
response = await get_answer(msg.query)
ctx.logger.info(response)
except Exception as err:
ctx.logger.error(err)
await ctx.send(
sender,
ErrorMessage(
error=str(err)
),
)
return
await ctx.send(
sender, UAgentResponse(message=response, type=UAgentResponseType.FINAL)
)
agent.include(toolhouseai_proto, publish_manifest=True)
if __name__ == "__main__":
agent.run()