-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathscript_skill_agent.py
More file actions
149 lines (131 loc) · 5.51 KB
/
script_skill_agent.py
File metadata and controls
149 lines (131 loc) · 5.51 KB
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
"""
Demonstrates BuiltinToolsSkill installation and tool usage lifecycle.
This example shows how to:
1. Initialize a World and an agent entity.
2. Install the BuiltinToolsSkill using SkillManager.
3. Configure the agent with Reasoning and ToolExecution systems.
4. Run a loop where the agent uses read_file, write_file, and glob.
"""
import asyncio
import os
import tempfile
from pathlib import Path
from ecs_agent import BuiltinToolsSkill, SkillManager
from ecs_agent.components import ConversationComponent, LLMComponent
from ecs_agent.core import Runner, World
from ecs_agent.providers import FakeModel, Model
from ecs_agent.providers.config import ApiFormat
from ecs_agent.systems.reasoning import ReasoningSystem
from ecs_agent.systems.tool_execution import ToolExecutionSystem
from ecs_agent.types import CompletionResult, Message, ToolCall
async def main() -> None:
# Set up a temporary workspace for file operations
with tempfile.TemporaryDirectory() as tmp_dir:
workspace = Path(tmp_dir)
test_file = workspace / "hello.txt"
test_file.write_text("Hello from the workspace!", encoding="utf-8")
world = World()
agent = world.create_entity()
# 1. Setup Provider (Use OpenAI if key is present, otherwise Fake)
api_key = os.getenv("LLM_API_KEY")
if api_key:
model = Model(os.getenv("LLM_MODEL", "qwen3.5-flash"), base_url=os.getenv("LLM_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"), api_key=api_key, api_format=ApiFormat.OPENAI_CHAT_COMPLETIONS)
else:
# Fake responses for the demo
model = FakeModel(
responses=[
# First response: call read_file
CompletionResult(
message=Message(
role="assistant",
content="Let me check the file.",
tool_calls=[
ToolCall(
id="call_1",
name="read_file",
arguments={"file_path": "hello.txt"},
)
],
)
),
# Second response: after seeing file content, write to it
CompletionResult(
message=Message(
role="assistant",
content="I will update the file.",
tool_calls=[
ToolCall(
id="call_2",
name="write_file",
arguments={"file_path": "hello.txt", "content": "Updated content!"},
)
],
)
),
# Third response: call glob to list workspace files
CompletionResult(
message=Message(
role="assistant",
content="Let me check what files are in the workspace.",
tool_calls=[
ToolCall(
id="call_3",
name="glob",
arguments={"pattern": "*.txt", "base_path": "."},
)
],
)
),
# Fourth response: finish
CompletionResult(
message=Message(
role="assistant", content="All done! I've updated the file."
)
),
]
)
# 2. Add components
world.add_component(
agent,
LLMComponent(
model=model,
system_prompt="You are a file manager.",
),
)
world.add_component(
agent,
ConversationComponent(
messages=[
Message(
role="user",
content="Read hello.txt and then change its content.",
)
]
),
)
# 3. Install Skills
manager = SkillManager()
# Bind workspace_root so the LLM doesn't need to provide it.
skill = BuiltinToolsSkill()
skill.bind_workspace(str(workspace))
manager.install(world, agent, skill)
# Register systems
world.register_system(ReasoningSystem(), priority=0)
world.register_system(ToolExecutionSystem(), priority=5)
# 4. Run the agent
print(f"Starting agent in workspace: {workspace}")
runner = Runner()
await runner.run(world, max_ticks=10)
# Verify results
final_content = test_file.read_text(encoding="utf-8")
print(f"Final file content: {final_content}")
# List installed skills
skills = manager.list_skills(world, agent)
print(f"Installed skills: {[s.name for s in skills]}")
# Demo uninstallation
manager.uninstall(world, agent, "builtin-tools")
print("Uninstalled builtin-tools skill.")
remaining_skills = manager.list_skills(world, agent)
print(f"Remaining skills: {[s.name for s in remaining_skills]}")
if __name__ == "__main__":
asyncio.run(main())