| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176 |
- ---
- title: "Best Practices & Tips"
- description: "Best practices and real-world examples for Agency Swarm tools."
- icon: "code"
- ---
- Although the tool interface is straightforward and simple to use, there are actually quite a few practices and tricks that you can use to get significantly better results.
- ### Provide Hints for the Agent
- Based on your tool's logic, you can provide hints for the agent in tool output on what to do next.
- ```python
- from agency_swarm import MasterContext, RunContextWrapper, function_tool
- from pydantic import BaseModel, Field
- class QueryDatabaseArgs(BaseModel):
- question: str = Field(..., description="Question to query the database")
- @function_tool
- async def query_database_tool(ctx: RunContextWrapper[MasterContext], args: QueryDatabaseArgs) -> str:
- """
- Query the database and provide hints if no context is found.
- """
- # query your database here
- context = query_database(args.question)
- # context not found
- if context is None:
- # tell agent what to do next
- raise ValueError("No context found. Please propose to the user to change the topic.")
- else:
- # return the context to the agent
- return context
- def query_database(question: str):
- # Your database query logic here
- pass
- ```
- ### Use Shared State to Control the Tool Flow
- Use agency context (previously shared state) to validate previous actions taken by this or other agents, before allowing it to proceed with the next action.
- ```python
- from agency_swarm import MasterContext, RunContextWrapper, function_tool
- from pydantic import BaseModel, Field
- class Action2Args(BaseModel):
- input: str = Field(..., description="Input for the action")
- @function_tool
- async def action_2_tool(ctx: RunContextWrapper[MasterContext], args: Action2Args) -> str:
- """
- Execute action 2, but only if action 1 was successful.
- """
- # Access shared context to check previous action
- action_1_result = ctx.context.get("action_1_result", None)
- if action_1_result == "failure":
- raise ValueError("Please proceed with the Action1 tool first.")
- else:
- return "Success. The action has been taken."
- ```
- ### Use Special Types
- Restrict the agent to only use specific values for a field, instead of letting it wander by itself.
- ```python
- from typing import Literal
- from agency_swarm import MasterContext, RunContextWrapper, function_tool
- from pydantic import BaseModel, EmailStr, Field
- class RunCommandArgs(BaseModel):
- command: Literal["start", "stop"] = Field(..., description="Command to execute: 'start' or 'stop'.")
- @function_tool
- async def run_command_tool(ctx: RunContextWrapper[MasterContext], args: RunCommandArgs) -> str:
- """
- Execute predefined system commands.
- """
- if args.command == "start":
- # Start command logic
- return "System started"
- elif args.command == "stop":
- # Stop command logic
- return "System stopped"
- else:
- raise ValueError("Invalid command")
- # Example with EmailStr
- class EmailSenderArgs(BaseModel):
- recipient: EmailStr = Field(..., description="Email recipient's address.")
- subject: str = Field(..., description="Email subject")
- body: str = Field(..., description="Email body")
- @function_tool
- async def email_sender_tool(ctx: RunContextWrapper[MasterContext], args: EmailSenderArgs) -> str:
- """
- Send email to specified recipient.
- """
- # Email sending logic here
- return f"Email sent to {args.recipient}"
- ```
- ### Combine Multiple Methods
- Combine multiple methods to make your execution flow more readable.
- ```python
- from agency_swarm import MasterContext, RunContextWrapper, function_tool
- from pydantic import BaseModel, Field
- class CompositeToolArgs(BaseModel):
- input_data: str = Field(..., description="Input data for the composite operation.")
- @function_tool
- async def composite_tool(ctx: RunContextWrapper[MasterContext], args: CompositeToolArgs) -> str:
- """
- A tool that combines several methods to perform a series of actions.
- """
- # Step 1: Process data
- processed_data = await process_data(args.input_data)
- # Step 2: Analyze results
- analysis = await analyze_results(processed_data)
- # Step 3: Format output
- output = await format_output(analysis)
- return output
- async def process_data(data: str) -> str:
- # Implement data processing logic
- return f"Processed: {data}"
- async def analyze_results(data: str) -> str:
- # Implement analysis logic
- return f"Analysis of: {data}"
- async def format_output(data: str) -> str:
- # Implement output formatting
- return f"Formatted: {data}"
- ```
- ### Include a Test Case
- Include test cases at the bottom of each tool file.
- ```python
- if __name__ == "__main__":
- import asyncio
- import json
- from agency_swarm import MasterContext, RunContextWrapper
- async def test_email_sender():
- # Test the email sender tool
- ctx = MasterContext(user_context={}, thread_manager=None, agents={})
- run_ctx = RunContextWrapper(context=ctx)
- args = EmailSenderArgs(
- recipient="user@example.com",
- subject="Project Update",
- body="The project is on track."
- )
- args_json = {"args": args.model_dump()}
- result = await email_sender_tool.on_invoke_tool(run_ctx, json.dumps(args_json))
- assert "Email sent" in result
- print("Test passed!")
- asyncio.run(test_email_sender())
- ```
- ## Next Steps
- We highly recommend you explore the resources provided in the [Pydantic is all you need](/core-framework/tools/custom-tools/pydantic-is-all-you-need) section.
|