Spaces:
Sleeping
Sleeping
File size: 3,600 Bytes
116524e | 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 | #!/usr/bin/env python3
"""Compose a custom ACE pipeline from individual steps.
Demonstrates:
1. Pipeline composition with a single import line from ace
2. Adding a custom step to the pipeline
3. Inspecting runner presets via build_steps()
4. Running the pipeline directly via ACERunner
Requires:
pip install ace-framework
export OPENAI_API_KEY=... # or any LiteLLM-supported provider
"""
from __future__ import annotations
from ace import (
# Pipeline engine
Pipeline,
StepProtocol,
# ACE context
ACEStepContext,
ACERunner,
# Roles
Agent,
Reflector,
SkillManager,
# Steps
AgentStep,
EvaluateStep,
learning_tail,
# Types
ACE,
LiteLLMClient,
Sample,
Skillbook,
SimpleEnvironment,
)
# ------------------------------------------------------------------
# 1. Custom step — print the agent's answer between execute and learn
# ------------------------------------------------------------------
class LogAnswerStep:
"""A custom step that logs the agent's output before learning."""
requires = frozenset({"agent_output"})
provides = frozenset()
def __call__(self, ctx: ACEStepContext) -> ACEStepContext:
print(f" -> Agent answered: {ctx.agent_output.final_answer}")
return ctx
# ------------------------------------------------------------------
# 2. Compose a custom pipeline
# ------------------------------------------------------------------
MODEL = "gpt-4o-mini"
llm = LiteLLMClient(model=MODEL)
skillbook = Skillbook()
pipe = Pipeline(
[
AgentStep(Agent(llm), skillbook),
EvaluateStep(SimpleEnvironment()),
LogAnswerStep(), # <-- custom step injected here
*learning_tail(Reflector(llm), SkillManager(llm), skillbook),
]
)
print(f"Custom pipeline: {len(pipe._steps)} steps")
print(f" requires: {pipe.requires}")
print(f" provides: {pipe.provides}")
# ------------------------------------------------------------------
# 3. Run using ACERunner
# ------------------------------------------------------------------
runner = ACERunner(pipeline=pipe, skillbook=skillbook)
samples = [
Sample(question="What is 2+2?", context="", ground_truth="4"),
Sample(question="Capital of France?", context="", ground_truth="Paris"),
]
# Note: ACERunner._build_context is abstract, so we use ACE which
# provides _build_context for Sample objects.
runner_ace = ACE(pipeline=pipe, skillbook=skillbook)
results = runner_ace.run(samples, epochs=1)
print(f"\nResults: {len(results)} samples processed")
print(f"Skills learned: {len(skillbook.skills())}")
# ------------------------------------------------------------------
# 4. Inspect and modify a runner's default steps with build_steps()
# ------------------------------------------------------------------
print("\n--- Inspecting ACE.build_steps() ---")
default_steps = ACE.build_steps(
agent=Agent(llm),
reflector=Reflector(llm),
skill_manager=SkillManager(llm),
environment=SimpleEnvironment(),
skillbook=Skillbook(),
)
for i, step in enumerate(default_steps):
print(f" [{i}] {type(step).__name__}")
# Modify: insert LogAnswerStep after EvaluateStep
default_steps.insert(2, LogAnswerStep())
print(f"\nAfter inserting LogAnswerStep: {len(default_steps)} steps")
# Build a new pipeline from the modified steps
modified_pipe = Pipeline(default_steps)
print(f"Modified pipeline ready with {len(modified_pipe._steps)} steps")
|