best-practices.mdx 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
  1. ---
  2. title: "Best Practices & Tips"
  3. description: "Best practices and real-world examples for Agency Swarm tools."
  4. icon: "code"
  5. ---
  6. 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.
  7. ### Provide Hints for the Agent
  8. Based on your tool's logic, you can provide hints for the agent in tool output on what to do next.
  9. ```python
  10. from agency_swarm import MasterContext, RunContextWrapper, function_tool
  11. from pydantic import BaseModel, Field
  12. class QueryDatabaseArgs(BaseModel):
  13. question: str = Field(..., description="Question to query the database")
  14. @function_tool
  15. async def query_database_tool(ctx: RunContextWrapper[MasterContext], args: QueryDatabaseArgs) -> str:
  16. """
  17. Query the database and provide hints if no context is found.
  18. """
  19. # query your database here
  20. context = query_database(args.question)
  21. # context not found
  22. if context is None:
  23. # tell agent what to do next
  24. raise ValueError("No context found. Please propose to the user to change the topic.")
  25. else:
  26. # return the context to the agent
  27. return context
  28. def query_database(question: str):
  29. # Your database query logic here
  30. pass
  31. ```
  32. ### Use Shared State to Control the Tool Flow
  33. 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.
  34. ```python
  35. from agency_swarm import MasterContext, RunContextWrapper, function_tool
  36. from pydantic import BaseModel, Field
  37. class Action2Args(BaseModel):
  38. input: str = Field(..., description="Input for the action")
  39. @function_tool
  40. async def action_2_tool(ctx: RunContextWrapper[MasterContext], args: Action2Args) -> str:
  41. """
  42. Execute action 2, but only if action 1 was successful.
  43. """
  44. # Access shared context to check previous action
  45. action_1_result = ctx.context.get("action_1_result", None)
  46. if action_1_result == "failure":
  47. raise ValueError("Please proceed with the Action1 tool first.")
  48. else:
  49. return "Success. The action has been taken."
  50. ```
  51. ### Use Special Types
  52. Restrict the agent to only use specific values for a field, instead of letting it wander by itself.
  53. ```python
  54. from typing import Literal
  55. from agency_swarm import MasterContext, RunContextWrapper, function_tool
  56. from pydantic import BaseModel, EmailStr, Field
  57. class RunCommandArgs(BaseModel):
  58. command: Literal["start", "stop"] = Field(..., description="Command to execute: 'start' or 'stop'.")
  59. @function_tool
  60. async def run_command_tool(ctx: RunContextWrapper[MasterContext], args: RunCommandArgs) -> str:
  61. """
  62. Execute predefined system commands.
  63. """
  64. if args.command == "start":
  65. # Start command logic
  66. return "System started"
  67. elif args.command == "stop":
  68. # Stop command logic
  69. return "System stopped"
  70. else:
  71. raise ValueError("Invalid command")
  72. # Example with EmailStr
  73. class EmailSenderArgs(BaseModel):
  74. recipient: EmailStr = Field(..., description="Email recipient's address.")
  75. subject: str = Field(..., description="Email subject")
  76. body: str = Field(..., description="Email body")
  77. @function_tool
  78. async def email_sender_tool(ctx: RunContextWrapper[MasterContext], args: EmailSenderArgs) -> str:
  79. """
  80. Send email to specified recipient.
  81. """
  82. # Email sending logic here
  83. return f"Email sent to {args.recipient}"
  84. ```
  85. ### Combine Multiple Methods
  86. Combine multiple methods to make your execution flow more readable.
  87. ```python
  88. from agency_swarm import MasterContext, RunContextWrapper, function_tool
  89. from pydantic import BaseModel, Field
  90. class CompositeToolArgs(BaseModel):
  91. input_data: str = Field(..., description="Input data for the composite operation.")
  92. @function_tool
  93. async def composite_tool(ctx: RunContextWrapper[MasterContext], args: CompositeToolArgs) -> str:
  94. """
  95. A tool that combines several methods to perform a series of actions.
  96. """
  97. # Step 1: Process data
  98. processed_data = await process_data(args.input_data)
  99. # Step 2: Analyze results
  100. analysis = await analyze_results(processed_data)
  101. # Step 3: Format output
  102. output = await format_output(analysis)
  103. return output
  104. async def process_data(data: str) -> str:
  105. # Implement data processing logic
  106. return f"Processed: {data}"
  107. async def analyze_results(data: str) -> str:
  108. # Implement analysis logic
  109. return f"Analysis of: {data}"
  110. async def format_output(data: str) -> str:
  111. # Implement output formatting
  112. return f"Formatted: {data}"
  113. ```
  114. ### Include a Test Case
  115. Include test cases at the bottom of each tool file.
  116. ```python
  117. if __name__ == "__main__":
  118. import asyncio
  119. import json
  120. from agency_swarm import MasterContext, RunContextWrapper
  121. async def test_email_sender():
  122. # Test the email sender tool
  123. ctx = MasterContext(user_context={}, thread_manager=None, agents={})
  124. run_ctx = RunContextWrapper(context=ctx)
  125. args = EmailSenderArgs(
  126. recipient="user@example.com",
  127. subject="Project Update",
  128. body="The project is on track."
  129. )
  130. args_json = {"args": args.model_dump()}
  131. result = await email_sender_tool.on_invoke_tool(run_ctx, json.dumps(args_json))
  132. assert "Email sent" in result
  133. print("Test passed!")
  134. asyncio.run(test_email_sender())
  135. ```
  136. ## Next Steps
  137. 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.