Planning

A big piece in chain of thought reasoning is planning. The zyx library comes with a plan() function, allowing helpful utility for planning for agentic tasks.

---

Tree of Thought Planning

Paper

from zyx import plan
from pydantic import BaseModel
from typing import List

goal = "Create a marketing strategy for a new smartphone"
tot_plan = plan(
    goal,
    steps=5,
    verbose = True
)

print(tot_plan)

Output

Plan(
    tasks=[
        Task(description='### Actionable Tasks for Integrated Multi-Channel Marketing Campaign', details=None),
        Task(description='1. **Develop a Comprehensive Marketing Plan:**', details=None),
        Task(
            description='- **Define Objectives:** Establish clear goals for the campaign, such as increasing brand awareness, driving traffic
to the website, or achieving a specific sales target.',
            details=None
        ),
        Task(
            description='- **Identify Target Audience:** Conduct research to define the target audience segments, including demographics,
interests, and behaviors.',
            details=None
        ),
        Task(
            description='- **Budget Allocation:** Determine the budget for each channel (social media, email, PPC, etc.) and allocate
resources accordingly.',
            details=None
        ),
        Task(
            description="- **Key Messages:** Craft core messages that resonate with the target audience and align with the brand's voice.",
            details=None
        ),
        Task(
            description='- **Timeline:** Create a timeline for the campaign, including key milestones and deadlines for each phase of the
marketing activities.',
            details=None
        ),
        Task(description='2. **Create Engaging Content:**', details=None),
        Task(
            description='- **Content Calendar:** Develop a content calendar that outlines what content will be published on which channels and
when.',
            details=None
        ),
        Task(description='- **Tailored Content Creation:** Produce high-quality content tailored for each platform, such as:', details=None),
        Task(
            description='- **Social Media Posts:** Eye-catching graphics and engaging captions for platforms like Instagram, Facebook, and
Twitter.',
            details=None
        ),
        Task(
            description="- **Email Newsletters:** Informative and visually appealing emails that highlight the smartphone's features and
promotions.",
            details=None
        ),
        Task(
            description="- **Blog Articles:** In-depth articles that provide insights into the smartphone's technology, benefits, and user
experiences.",
            details=None
        ),
        Task(
            description="- **Video Ads:** Create short, engaging videos showcasing the smartphone's features and user testimonials.",
            details=None
        ),
        Task(description='3. **Leverage Influencer Partnerships:**', details=None),
        Task(
            description="- **Identify Influencers:** Research and compile a list of influencers who align with the brand's values and have a
following that matches the target audience.",
            details=None
        ),
        Task(
            description='- **Outreach Strategy:** Develop a strategy for reaching out to influencers, including personalized messages and
collaboration proposals.',
            details=None
        ),
        Task(
            description='- **Content Collaboration:** Work with influencers to create authentic content that showcases the smartphone, such as
unboxing videos, reviews, or lifestyle posts.',
            details=None
        ),
        Task(
            description='- **Track Engagement:** Monitor the performance of influencer content to assess reach, engagement, and conversion
rates.',
            details=None
        ),
        Task(description='4. **Implement Tracking and Analytics:**', details=None),
        Task(
            description='- **Set Up Tracking Tools:** Utilize tools like Google Analytics, social media insights, and email marketing
analytics to track campaign performance.',
            details=None
        ),
        Task(
            description='- **Define KPIs:** Establish key performance indicators (KPIs) to measure success, such as website traffic,
conversion rates, and social media engagement.',
            details=None
        ),
        Task(
            description='- **Real-Time Monitoring:** Implement real-time monitoring to assess the effectiveness of each channel and make
adjustments as needed.',
            details=None
        ),
        Task(
            description='- **Post-Campaign Analysis:** After the campaign, conduct a thorough analysis of the data to evaluate what worked
well and what can be improved for future campaigns.',
            details=None
        ),
        Task(description='5. **Launch and Promote the Campaign:**', details=None),
        Task(
            description='- **Coordinated Launch:** Ensure that all channels are prepared for the launch, with content scheduled and ready to
go live simultaneously.',
            details=None
        ),
        Task(
            description='- **Engagement Strategies:** Implement strategies to engage the audience during the launch, such as live Q&A
sessions, giveaways, or contests.',
            details=None
        ),
        Task(
            description="- **Consistent Messaging:** Maintain consistent messaging across all channels to reinforce the campaign's key
messages and brand identity.",
            details=None
        ),
        Task(
            description='- **Follow-Up Promotions:** Plan follow-up promotions or content to sustain engagement and interest after the initial
launch.',
            details=None
        )
    ]
)

---

Planning with a Custom BaseModel Input

import zyx
from pydantic import BaseModel
from typing import List

class ResearchTask(BaseModel):
    topic: str
    resources: List[str]
    estimated_time: int

plan_model_tot = zyx.plan(
    ResearchTask,
    steps=4,
    verbose=True
)

print(plan_model_tot)

Output

ResearchTaskPlan(
    tasks=[
        ResearchTask(
            topic='Understanding Machine Learning',
            resources=[
                "Textbook: 'Pattern Recognition and Machine Learning'",
                "Online Course: 'Machine Learning by Stanford'",
                "Research Paper: 'A survey of machine learning'"
            ],
            estimated_time=12
        ),
        ResearchTask(
            topic='Exploring Neural Networks',
            resources=[
                "Textbook: 'Deep Learning' by Ian Goodfellow",
                'Video Lecture Series on YouTube',
                "Research Paper: 'Neural Networks for Machine Learning'"
            ],
            estimated_time=15
        ),
        ResearchTask(
            topic='Data Preprocessing Techniques',
            resources=[
                "Online Article: 'A Comprehensive Guide to Data Preprocessing'",
                'Video Tutorial on Data Cleaning',
                "Research Paper: 'Data Preprocessing for Machine Learning'"
            ],
            estimated_time=8
        ),
        ResearchTask(
            topic='Evaluation Metrics in Machine Learning',
            resources=[
                "Online Course: 'Evaluation Metrics for ML Models'",
                "Textbook: 'The Elements of Statistical Learning'",
                "Research Paper: 'Statistical Methods for Evaluating Learning Algorithms'"
            ],
            estimated_time=10
        )
    ]
)

---

API Reference

Generates a plan or batch of plans based on the input using the Tree of Thoughts method.

Example

>>> plan(
    input="Create a marketing strategy for a new smartphone",
    steps=5,
    verbose=True
)

Parameters:

Name	Type	Description	Default
input	Union[str, Type[BaseModel]]	The input can be either a string describing the task or a Pydantic model class.	required
instructions	Optional[str]	Additional instructions for the planning process.	None
process	Literal['single', 'batch']	Process can be either "single" or "batch".	'single'
n	int	Number of plans to generate.	1
batch_size	int	Number of plans to generate in a single batch.	3
steps	int	Number of steps per plan.	5
model	str	The model to use for the planning process.	'gpt-4o-mini'
api_key	Optional[str]	The API key to use for the planning process.	None
base_url	Optional[str]	The base URL to use for the planning process.	None
organization	Optional[str]	The organization to use for the planning process.	None
temperature	Optional[float]	The temperature to use for the planning process.	None
mode	InstructorMode	The mode to use for the planning process.	'markdown_json_mode'
max_retries	int	The maximum number of retries to use for the planning process.	3
client	Optional[Literal['openai', 'litellm']]	The client to use for the planning process.	None
verbose	bool	Whether to print the planning process to the console.	False

Returns:

Type	Description
Union[Plan, List[Plan], Any, List[Any]]	Union[Plan, List[Plan], Any, List[Any]]: The plan or batch of plans.

Source code in zyx/resources/completions/agents/plan.py

def plan(
    input: Union[str, Type[BaseModel]],
    instructions: Optional[str] = None,
    process: Literal["single", "batch"] = "single",
    n: int = 1,
    batch_size: int = 3,
    steps: int = 5,
    model: str = "gpt-4o-mini",
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
    organization: Optional[str] = None,
    temperature: Optional[float] = None,
    mode: InstructorMode = "markdown_json_mode",
    max_retries: int = 3,
    client: Optional[Literal["openai", "litellm"]] = None,
    verbose: bool = False,
) -> Union[Plan, List[Plan], Any, List[Any]]:
    """
    Generates a plan or batch of plans based on the input using the Tree of Thoughts method.

    Example:
        ```python
        >>> plan(
            input="Create a marketing strategy for a new smartphone",
            steps=5,
            verbose=True
        )
        ```

    Args:
        input (Union[str, Type[BaseModel]]): The input can be either a string describing the task or a Pydantic model class.
        instructions (Optional[str]): Additional instructions for the planning process.
        process (Literal["single", "batch"]): Process can be either "single" or "batch".
        n (int): Number of plans to generate.
        batch_size (int): Number of plans to generate in a single batch.
        steps (int): Number of steps per plan.
        model (str): The model to use for the planning process.
        api_key (Optional[str]): The API key to use for the planning process.
        base_url (Optional[str]): The base URL to use for the planning process.
        organization (Optional[str]): The organization to use for the planning process.
        temperature (Optional[float]): The temperature to use for the planning process.
        mode (InstructorMode): The mode to use for the planning process.
        max_retries (int): The maximum number of retries to use for the planning process.
        client (Optional[Literal["openai", "litellm"]]): The client to use for the planning process.
        verbose (bool): Whether to print the planning process to the console.

    Returns:
        Union[Plan, List[Plan], Any, List[Any]]: The plan or batch of plans.
    """

    if verbose:
        logger.info(f"Generating {n} plan(s) using Tree of Thoughts method")
        logger.info(f"Using model: {model}")
        logger.info(f"Number of steps per plan: {steps}")
        logger.info(f"Process: {process}")
        logger.info(f"Batch size: {batch_size}")

    completion_client = Client(
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        provider=client,
        verbose=verbose,
    )

    if isinstance(input, str):
        system_message = _get_string_system_message(input, steps)
        response_model = Plan
    elif isinstance(input, type) and issubclass(input, BaseModel):
        system_message = _get_model_system_message(input, steps)
        response_model = create_model(f"{input.__name__}Plan", tasks=(List[input], ...))
    else:
        raise ValueError("Input must be either a string or a Pydantic model class.")

    user_message = (
        instructions if instructions else f"Generate a plan with {steps} steps."
    )

    if process == "single" or n == 1:
        result = completion_client.completion(
            messages=[
                {"role": "system", "content": system_message},
                {"role": "user", "content": user_message},
            ],
            model=model,
            response_model=response_model,
            mode=mode,
            max_retries=max_retries,
            temperature=temperature,
        )
        return result
    else:  # batch process
        batch_response_model = create_model(
            "ResponseModel", items=(List[response_model], ...)
        )
        results = []
        for i in range(0, n, batch_size):
            batch_n = min(batch_size, n - i)
            batch_message = f"Generate {batch_n} plans, each with {steps} steps."
            if results:
                batch_message += f"\nPreviously generated plans: {results[-3:]}\nEnsure these new plans are different."

            result = completion_client.completion(
                messages=[
                    {"role": "system", "content": system_message},
                    {"role": "user", "content": batch_message},
                ],
                model=model,
                response_model=batch_response_model,
                mode=mode,
                max_retries=max_retries,
                temperature=temperature,
            )

            results.extend(result.items)

        return results