import dspy
from roma_dspy import Aggregator, Atomizer, Executor, Planner, Verifier, SubTask
from roma_dspy.types import TaskType
# Optional tool that the Executor may call
def get_weather(city: str) -> str:
"""Return a canned weather report for the city."""
return f"The weather in {city} is sunny."
# Executor geared toward ReAct with a Fireworks model
executor_lm = dspy.LM(
"fireworks_ai/accounts/fireworks/models/kimi-k2-instruct-0905",
temperature=0.7,
cache=True,
)
executor = Executor(
lm=executor_lm,
prediction_strategy="react",
tools=[get_weather],
context_defaults={"track_usage": True},
)
# Atomizer decides when to branch into planning
atomizer = Atomizer(
lm=dspy.LM("openrouter/google/gemini-2.5-flash", temperature=0.6, cache=False),
prediction_strategy="cot",
context_defaults={"track_usage": True},
)
# Planner produces executable subtasks for non-atomic goals
planner = Planner(
lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.85, cache=True),
prediction_strategy="cot",
context_defaults={"track_usage": True},
)
aggregator = Aggregator(
lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.65),
prediction_strategy="cot",
)
verifier = Verifier(
lm=dspy.LM("openrouter/openai/gpt-4o-mini", temperature=0.0),
)
def run_pipeline(goal: str) -> str:
atomized = atomizer.forward(goal)
if atomized.is_atomic or atomized.node_type.is_execute:
execution = executor.forward(goal)
candidate = execution.output
else:
plan = planner.forward(goal)
results = []
for idx, subtask in enumerate(plan.subtasks, start=1):
execution = executor.forward(subtask.goal)
results.append(
SubTask(
goal=subtask.goal,
task_type=subtask.task_type,
dependencies=subtask.dependencies,
)
)
aggregated = aggregator.forward(goal, results)
candidate = aggregated.synthesized_result
verdict = verifier.forward(goal, candidate)
if verdict.verdict:
return candidate
return f"Verifier flagged the output: {verdict.feedback or 'no feedback returned'}"
print(run_pipeline("Plan a weekend in Barcelona and include a packing list."))
Open AGI 峰会
一个供开发者、创始人及研究人员探讨开源 AI 未来的空间。
汇聚拥有共同信念的研究人员、开发者及社区——即 AGI 必须保持开放，并让每个人都能触达。
.webp)
.webp)
.webp)
.webp)
.webp)
.webp)
