KeyArgo
/
smolagents
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Simplify managed agents (#484) 

* Simplify managed agents
This commit is contained in:
Aymeric Roucher 2025-02-03 21:19:54 +01:00 committed by GitHub
parent dd84e7333d
commit d69ae028fa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 72 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
docs/source/en/examples/multiagents.md
View File

@ -19,7 +19,7 @@ rendered properly in your Markdown viewer.
In this notebook we will make a **multi-agent web browser: an agentic system with several agents collaborating to solve problems using the web!**
It will be a simple hierarchy, using a `ManagedAgent` object to wrap the managed web search agent:
It will be a simple hierarchy:
```
+----------------+
@ -28,15 +28,12 @@ It will be a simple hierarchy, using a `ManagedAgent` object to wrap the managed
|
_______________|______________
| |
Code interpreter +--------------------------------+
tool | Managed agent |
| +------------------+ |
| | Web Search agent | |
| +------------------+ |
| | | |
| Web Search tool | |
| Visit webpage tool |
+--------------------------------+
Code Interpreter +------------------+
tool | Web Search agent |
+------------------+
| |
Web Search tool |
Visit webpage tool
```
Let's set up this system.
@ -127,7 +124,6 @@ from smolagents import (
CodeAgent,
ToolCallingAgent,
HfApiModel,
ManagedAgent,
DuckDuckGoSearchTool,
LiteLLMModel,
)
@ -138,20 +134,14 @@ web_agent = ToolCallingAgent(
tools=[DuckDuckGoSearchTool(), visit_webpage],
model=model,
max_steps=10,
)
```
We then wrap this agent into a `ManagedAgent` that will make it callable by its manager agent.
```py
managed_web_agent = ManagedAgent(
agent=web_agent,
name="search",
description="Runs web searches for you. Give it your query as an argument.",
)
```
Finally we create a manager agent, and upon initialization we pass our managed agent to it in its `managed_agents` argument.
Note that we gave this agent attributes `name` and `description`, mandatory attributes to make this agent callable by its manager agent.
Then we create a manager agent, and upon initialization we pass our managed agent to it in its `managed_agents` argument.
Since this agent is the one tasked with the planning and thinking, advanced reasoning will be beneficial, so a `CodeAgent` will be the best choice.
@ -161,7 +151,7 @@ Also, we want to ask a question that involves the current year and does addition
manager_agent = CodeAgent(
tools=[],
model=model,
managed_agents=[managed_web_agent],
managed_agents=[web_agent],
additional_authorized_imports=["time", "numpy", "pandas"],
)
```

2
docs/source/en/reference/agents.md
View File

@ -45,7 +45,7 @@ Both require arguments `model` and list of tools `tools` at initialization.
### ManagedAgent
[[autodoc]] ManagedAgent
_This class is deprecated since 1.8.0: now you simply need to pass attributes `name` and `description` to a normal agent to make it callable by a manager agent._
### stream_to_gradio

10
docs/source/en/reference/models.md
View File

@ -61,7 +61,7 @@ from smolagents import TransformersModel
model = TransformersModel(model_id="HuggingFaceTB/SmolLM-135M-Instruct")
print(model([{"role": "user", "content": "Ok!"}], stop_sequences=["great"]))
print(model([{"role": "user", "content": [{"type": "text", "text": "Ok!"}]}], stop_sequences=["great"]))
```
```text
>>> What a
@ -80,9 +80,7 @@ The `HfApiModel` wraps huggingface_hub's [InferenceClient](https://huggingface.c
from smolagents import HfApiModel
messages = [
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing great. How can I help you today?"},
{"role": "user", "content": "No need to help, take it easy."},
{"role": "user", "content": [{"type": "text", "text": "Hello, how are you?"}]}
]
model = HfApiModel()
@ -102,9 +100,7 @@ You can pass kwargs upon model initialization that will then be used whenever us
from smolagents import LiteLLMModel
messages = [
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing great. How can I help you today?"},
{"role": "user", "content": "No need to help, take it easy."},
{"role": "user", "content": [{"type": "text", "text": "Hello, how are you?"}]}
]
model = LiteLLMModel("anthropic/claude-3-5-sonnet-latest", temperature=0.2, max_tokens=10)

11
docs/source/en/tutorials/inspect_runs.md
View File

@ -78,7 +78,6 @@ Then you can run your agents!
from smolagents import (
CodeAgent,
ToolCallingAgent,
ManagedAgent,
DuckDuckGoSearchTool,
VisitWebpageTool,
HfApiModel,
@ -86,19 +85,17 @@ from smolagents import (
model = HfApiModel()
agent = ToolCallingAgent(
search_agent = ToolCallingAgent(
tools=[DuckDuckGoSearchTool(), VisitWebpageTool()],
model=model,
)
managed_agent = ManagedAgent(
agent=agent,
name="managed_agent",
name="search_agent",
description="This is an agent that can do web search.",
)
manager_agent = CodeAgent(
tools=[],
model=model,
managed_agents=[managed_agent],
managed_agents=[search_agent],
)
manager_agent.run(
"If the US keeps its 2024 growth rate, how many years will it take for the GDP to double?"

2
docs/source/zh/reference/agents.md
View File

@ -47,7 +47,7 @@ Both require arguments `model` and list of tools `tools` at initialization.
### ManagedAgent
[[autodoc]] ManagedAgent
_This class is deprecated since 1.8.0: now you just need to pass name and description attributes to an agent to use it as a ManagedAgent._
### stream_to_gradio

2
examples/e2b_example.py
View File

@ -42,7 +42,7 @@ agent = CodeAgent(
)
agent.run(
"Return me an image of a cat. Directly use the image provided in your state.",
"Calculate how much is 2+2, then return me an image of a cat. Directly use the image provided in your state.",
additional_args={"cat_image": get_cat_image()},
) # Asking to directly return the image from state tests that additional_args are properly sent to server.

11
examples/inspect_runs.py → examples/inspect_multiagent_run.py
View File

@ -7,7 +7,6 @@ from smolagents import (
CodeAgent,
DuckDuckGoSearchTool,
HfApiModel,
ManagedAgent,
ToolCallingAgent,
VisitWebpageTool,
)
@ -23,18 +22,16 @@ SmolagentsInstrumentor().instrument(tracer_provider=trace_provider, skip_dep_che
# Then we run the agentic part!
model = HfApiModel()
agent = ToolCallingAgent(
search_agent = ToolCallingAgent(
tools=[DuckDuckGoSearchTool(), VisitWebpageTool()],
model=model,
)
managed_agent = ManagedAgent(
agent=agent,
name="managed_agent",
name="search_agent",
description="This is an agent that can do web search.",
)
manager_agent = CodeAgent(
tools=[],
model=model,
managed_agents=[managed_agent],
managed_agents=[search_agent],
)
manager_agent.run("If the US keeps it 2024 growth rate, how many years would it take for the GDP to double?")

88
src/smolagents/agents.py
View File

@ -140,6 +140,9 @@ class MultiStepAgent:
managed_agents (`list`, *optional*): Managed agents that the agent can call.
step_callbacks (`list[Callable]`, *optional*): Callbacks that will be called at each step.
planning_interval (`int`, *optional*): Interval at which the agent will run a planning step.
name (`str`, *optional*): Necessary for a managed agent only - the name by which this agent can be called.
description (`str`, *optional*): Necessary for a managed agent only - the description of this agent.
managed_agent_prompt (`str`, *optional*): Custom prompt for the managed agent. Defaults to None.
"""
def __init__(
@ -156,6 +159,9 @@ class MultiStepAgent:
managed_agents: Optional[List] = None,
step_callbacks: Optional[List[Callable]] = None,
planning_interval: Optional[int] = None,
name: Optional[str] = None,
description: Optional[str] = None,
managed_agent_prompt: Optional[str] = None,
):
if system_prompt is None:
system_prompt = CODE_SYSTEM_PROMPT
@ -172,9 +178,16 @@ class MultiStepAgent:
self.grammar = grammar
self.planning_interval = planning_interval
self.state = {}
self.name = name
self.description = description
self.managed_agent_prompt = managed_agent_prompt if managed_agent_prompt else MANAGED_AGENT_PROMPT
self.managed_agents = {}
if managed_agents is not None:
for managed_agent in managed_agents:
assert managed_agent.name and managed_agent.description, (
"All managed agents need both a name and a description!"
)
self.managed_agents = {agent.name: agent for agent in managed_agents}
for tool in tools:
@ -638,6 +651,22 @@ Now begin!""",
"""
self.memory.replay(self.logger, detailed=detailed)
def __call__(self, request, provide_run_summary=False, **kwargs):
"""Adds additional prompting for the managed agent, and runs it."""
full_task = self.managed_agent_prompt.format(name=self.name, task=request).strip()
output = self.run(full_task, **kwargs)
if provide_run_summary:
answer = f"Here is the final answer from your managed agent '{self.name}':\n"
answer += str(output)
answer += f"\n\nFor more detail, find below a summary of this agent's work:\nSUMMARY OF WORK FROM AGENT '{self.name}':\n"
for message in self.write_memory_to_messages(summary_mode=True):
content = message["content"]
answer += "\n" + truncate_content(str(content)) + "\n---"
answer += f"\nEND OF SUMMARY OF WORK FROM AGENT '{self.name}'."
return answer
else:
return output
class ToolCallingAgent(MultiStepAgent):
"""
@ -896,7 +925,7 @@ class CodeAgent(MultiStepAgent):
]
observation = "Execution logs:\n" + execution_logs
except Exception as e:
if "print_outputs" in self.python_executor.state:
if hasattr(self.python_executor, "state") and "print_outputs" in self.python_executor.state:
execution_logs = self.python_executor.state["print_outputs"]
if len(execution_logs) > 0:
execution_outputs_console = [
@ -928,59 +957,4 @@ class CodeAgent(MultiStepAgent):
return output if is_final_answer else None
class ManagedAgent:
"""
ManagedAgent class that manages an agent and provides additional prompting and run summaries.
Args:
agent (`object`): The agent to be managed.
name (`str`): The name of the managed agent.
description (`str`): A description of the managed agent.
additional_prompting (`Optional[str]`, *optional*): Additional prompting for the managed agent. Defaults to None.
provide_run_summary (`bool`, *optional*): Whether to provide a run summary after the agent completes its task. Defaults to False.
managed_agent_prompt (`Optional[str]`, *optional*): Custom prompt for the managed agent. Defaults to None.
"""
def __init__(
self,
agent,
name,
description,
additional_prompting: Optional[str] = None,
provide_run_summary: bool = False,
managed_agent_prompt: Optional[str] = None,
):
self.agent = agent
self.name = name
self.description = description
self.additional_prompting = additional_prompting
self.provide_run_summary = provide_run_summary
self.managed_agent_prompt = managed_agent_prompt if managed_agent_prompt else MANAGED_AGENT_PROMPT
def write_full_task(self, task):
"""Adds additional prompting for the managed agent, like 'add more detail in your answer'."""
full_task = self.managed_agent_prompt.format(name=self.name, task=task)
if self.additional_prompting:
full_task = full_task.replace("\n{additional_prompting}", self.additional_prompting).strip()
else:
full_task = full_task.replace("\n{additional_prompting}", "").strip()
return full_task
def __call__(self, request, **kwargs):
full_task = self.write_full_task(request)
output = self.agent.run(full_task, **kwargs)
if self.provide_run_summary:
answer = f"Here is the final answer from your managed agent '{self.name}':\n"
answer += str(output)
answer += f"\n\nFor more detail, find below a summary of this agent's work:\nSUMMARY OF WORK FROM AGENT '{self.name}':\n"
for message in self.agent.write_memory_to_messages(summary_mode=True):
content = message["content"]
answer += "\n" + truncate_content(str(content)) + "\n---"
answer += f"\nEND OF SUMMARY OF WORK FROM AGENT '{self.name}'."
return answer
else:
return output
__all__ = ["ManagedAgent", "MultiStepAgent", "CodeAgent", "ToolCallingAgent", "AgentMemory"]
__all__ = ["MultiStepAgent", "CodeAgent", "ToolCallingAgent", "AgentMemory"]

10
src/smolagents/e2b_executor.py
View File

@ -45,9 +45,11 @@ class E2BExecutor:
"""Please install 'e2b' extra to use E2BExecutor: `pip install "smolagents[e2b]"`"""
)
self.logger.log("Initializing E2B executor, hold on...")
self.custom_tools = {}
self.final_answer = False
self.final_answer_pattern = re.compile(r"^final_answer\((.*)\)$")
self.final_answer_pattern = re.compile(r"final_answer\((.*?)\)")
self.sbx = Sandbox() # "qywp2ctmu2q7jzprcf4j")
# TODO: validate installing agents package or not
# print("Installing agents package on remote executor...")
@ -90,7 +92,7 @@ class E2BExecutor:
self.logger.log(tool_definition_execution.logs)
def run_code_raise_errors(self, code: str):
if self.final_answer_pattern.match(code):
if self.final_answer_pattern.search(code) is not None:
self.final_answer = True
execution = self.sbx.run_code(
code,
@ -152,7 +154,9 @@ locals().update({key: value for key, value in pickle_dict.items()})
]:
if getattr(result, attribute_name) is not None:
return getattr(result, attribute_name), execution_logs, self.final_answer
raise ValueError("No main result returned by executor!")
if self.final_answer:
raise ValueError("No main result returned by executor!")
return None, execution_logs, False
__all__ = ["E2BExecutor"]

2
src/smolagents/prompts.py
View File

@ -510,7 +510,7 @@ Your final_answer WILL HAVE to contain these parts:
Put all these in your final_answer tool, everything that you do not pass as an argument to final_answer will be lost.
And even if your task resolution is not successful, please return as much context as possible, so that your manager can act upon this feedback.
{{additional_prompting}}"""
"""
__all__ = [
"USER_PROMPT_PLAN_UPDATE",

23
tests/test_agents.py
View File

@ -25,7 +25,6 @@ from smolagents.agent_types import AgentImage, AgentText
from smolagents.agents import (
AgentMaxStepsError,
CodeAgent,
ManagedAgent,
MultiStepAgent,
ToolCall,
ToolCallingAgent,
@ -465,22 +464,20 @@ class AgentTests(unittest.TestCase):
assert res[0] == 0.5
def test_init_managed_agent(self):
agent = CodeAgent(tools=[], model=fake_code_functiondef)
managed_agent = ManagedAgent(agent, name="managed_agent", description="Empty")
assert managed_agent.name == "managed_agent"
assert managed_agent.description == "Empty"
agent = CodeAgent(tools=[], model=fake_code_functiondef, name="managed_agent", description="Empty")
assert agent.name == "managed_agent"
assert agent.description == "Empty"
def test_agent_description_gets_correctly_inserted_in_system_prompt(self):
agent = CodeAgent(tools=[], model=fake_code_functiondef)
managed_agent = ManagedAgent(agent, name="managed_agent", description="Empty")
managed_agent = CodeAgent(tools=[], model=fake_code_functiondef, name="managed_agent", description="Empty")
manager_agent = CodeAgent(
tools=[],
model=fake_code_functiondef,
managed_agents=[managed_agent],
)
assert "You can also give requests to team members." not in agent.system_prompt
assert "You can also give requests to team members." not in managed_agent.system_prompt
print("ok1")
assert "{{managed_agents_descriptions}}" not in agent.system_prompt
assert "{{managed_agents_descriptions}}" not in managed_agent.system_prompt
assert "You can also give requests to team members." in manager_agent.system_prompt
def test_code_agent_missing_import_triggers_advice_in_error_log(self):
@ -587,10 +584,6 @@ final_answer("Final report.")
tools=[],
model=managed_model,
max_steps=10,
)
managed_web_agent = ManagedAgent(
agent=web_agent,
name="search_agent",
description="Runs web searches for you. Give it your request as an argument. Make the request as detailed as needed, you can ask for thorough reports",
)
@ -598,7 +591,7 @@ final_answer("Final report.")
manager_code_agent = CodeAgent(
tools=[],
model=manager_model,
managed_agents=[managed_web_agent],
managed_agents=[web_agent],
additional_authorized_imports=["time", "numpy", "pandas"],
)
@ -608,7 +601,7 @@ final_answer("Final report.")
manager_toolcalling_agent = ToolCallingAgent(
tools=[],
model=manager_model,
managed_agents=[managed_web_agent],
managed_agents=[web_agent],
)
report = manager_toolcalling_agent.run("Fake question.")