from strands import Agent
from strands.models.bedrock import BedrockModel
from strands_tools import calculator

agent = Agent(system_prompt="You are a math agent", tools=[calculator])

message = "Calculate 125/5"
print(agent(message))

"""
I'll calculate 125/5 for you.

Tool #1: calculator

The result of 125/5 is **25**.The result of 125/5 is **25**.

"""

from langgraph.prebuilt import create_react_agent
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
import asyncio

@tool
def calculator(expression: str) -> str:
    """Evaluates a mathematical expression and returns the result.

    This tool takes a string containing a mathematical expression, evaluates it
    using Python's eval() function, and returns the result. It handles basic
    arithmetic operations, mathematical functions, and can work with complex
    expressions.

    Args:
        expression (str): A string containing a mathematical expression to evaluate,
                          such as "2 + 2", "5 * (3 + 2)", or "math.sin(0.5)".

    Returns:
        str: A string containing the result of the evaluation prefixed with "Result: ",
             or an error message if the evaluation fails.

    Raises:
        Exception: The function catches all exceptions during evaluation and returns
                   them as error messages in the returned string.

    Examples:
        >>> calculator("2 + 2")
        "Result: 4"
        >>> calculator("5 * (3 + 2)")
        "Result: 25"
        >>> calculator("10/0")
        "Error calculating expression: division by zero"
    """
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error calculating expression: {str(e)}"

llm = ChatBedrockConverse(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    region_name="us-east-1",
)
agent = create_react_agent(llm, tools=[calculator], prompt="You are a math agent")

async def run_agent(message: str):
    result = await agent.ainvoke({"messages": [HumanMessage(content=message)]})
    return result

if __name__ == "__main__":
    message = "Calculate 125/5"
    result = asyncio.run(run_agent(message))
    print(result)

from llama_index.core.agent.workflow import ReActAgent
from llama_index.llms.bedrock_converse import BedrockConverse
import asyncio

def calculator(expression: str) -> str:
    """Evaluates a mathematical expression and returns the result.

    This tool takes a string containing a mathematical expression, evaluates it
    using Python's eval() function, and returns the result. It handles basic
    arithmetic operations, mathematical functions, and can work with complex
    expressions.

    Args:
        expression (str): A string containing a mathematical expression to evaluate,
                          such as "2 + 2", "5 * (3 + 2)", or "math.sin(0.5)".

    Returns:
        str: A string containing the result of the evaluation.
    """
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error calculating expression: {str(e)}"

llm = BedrockConverse(
    model="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

agent = ReActAgent(
    name="math_agent",
    system_prompt="You are a math agent that can solve mathematical problems. Use the calculator tool to evaluate expressions.",
    tools=[calculator],
    llm=llm,
)

async def run_agent(message: str):
    """Run the agent with a given message.

    Args:
        message (str): The user's message to the agent

    Returns:
        The agent's response
    """
    response = await agent.run(message)
    return response

if __name__ == "__main__":
    message = "Calculate 125/5"
    result = asyncio.run(run_agent(message))
    print(result)
    print(result.tool_calls)
    
 """
 The result of 125 divided by 5 is 25.
[ToolCallResult(tool_name='calculator', tool_kwargs={'expression': '125/5'}, tool_id='16d0b75c-0a68-4f2e-a219-1165fc4ee2e8', tool_output=ToolOutput(blocks=[TextBlock(block_type='text', text='Result: 25.0')], tool_name='calculator', raw_input={'args': (), 'kwargs': {'expression': '125/5'}}, raw_output='Result: 25.0', is_error=False), return_direct=False)]
 """

from typing import Type

from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    expression: str = Field(..., description="The mathematical expression to evaluate.")

class CalculatorTool(BaseTool):
    name: str = "Calculator Tool"
    description: str = "Evaluates a mathematical expression and returns the result."
    args_schema: Type[BaseModel] = CalculatorInput

    def _run(self, expression: str) -> str:
        try:
            result = eval(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error calculating expression: {str(e)}"

def create_math_agent():

    llm = LLM(
        model="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
        aws_region_name="us-east-1",
    )

    math_agent = Agent(
        role="A skilled mathematician who can solve mathematical problems.",
        goal="Solve mathematical problems accurately and efficiently.",
        backstory="You are an expert mathematician with years of experience solving complex mathematical problems.",
        verbose=True,
        allow_delegation=False,
        tools=[CalculatorTool()],
        llm=llm,
    )

    return math_agent

def run_agent(message: str):
    math_agent = create_math_agent()

    task = Task(
        description=message,
        agent=math_agent,
        expected_output="The solution to the mathematical problem.",
    )

    crew = Crew(
        agents=[math_agent],
        tasks=[task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    return result

if __name__ == "__main__":
    message = "Calculate 125/5"
    result = run_agent(message)
    print(result)

from typing import Optional

from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

from self_evaluation_loop_flow.crews.shakespeare_crew.shakespeare_crew import (
    ShakespeareanXPostCrew,
)
from self_evaluation_loop_flow.crews.x_post_review_crew.x_post_review_crew import (
    XPostReviewCrew,
)
from dotenv import load_dotenv

load_dotenv()

class ShakespeareXPostFlowState(BaseModel):
    x_post: str = ""
    feedback: Optional[str] = None
    valid: bool = False
    retry_count: int = 0

class ShakespeareXPostFlow(Flow[ShakespeareXPostFlowState]):

    @start("retry")
    def generate_shakespeare_x_post(self):
        print("Generating Shakespearean X post")
        topic = "Flying cars"
        result = (
            ShakespeareanXPostCrew()
            .crew()
            .kickoff(inputs={"topic": topic, "feedback": self.state.feedback})
        )

        print("X post generated", result.raw)
        self.state.x_post = result.raw

    @router(generate_shakespeare_x_post)
    def evaluate_x_post(self):
        if self.state.retry_count > 3:
            return "max_retry_exceeded"

        result = XPostReviewCrew().crew().kickoff(inputs={"x_post": self.state.x_post})
        self.state.valid = result["valid"]
        self.state.feedback = result["feedback"]

        print("valid", self.state.valid)
        print("feedback", self.state.feedback)
        self.state.retry_count += 1

        if self.state.valid:
            return "complete"

        return "retry"

    @listen("complete")
    def save_result(self):
        print("X post is valid")
        print("X post:", self.state.x_post)

        with open("x_post.txt", "w") as file:
            file.write(self.state.x_post)

    @listen("max_retry_exceeded")
    def max_retry_exceeded_exit(self):
        print("Max retry count exceeded")
        print("X post:", self.state.x_post)
        print("Feedback:", self.state.feedback)

def kickoff():
    shakespeare_flow = ShakespeareXPostFlow()
    shakespeare_flow.kickoff()

def plot():
    shakespeare_flow = ShakespeareXPostFlow()
    shakespeare_flow.plot()

if __name__ == "__main__":
    kickoff()

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
    Context,
)

from llama_index.core.agent.workflow import ReActAgent
from llama_index.llms.bedrock_converse import BedrockConverse
from typing import Optional

from pydantic import BaseModel
import asyncio

class PostEvent(StartEvent):
    topic: str
    feedback: Optional[str] = None

class VerifyPostEvent(Event):
    post: str

class VerifyPostOutput(BaseModel):
    valid: bool
    feedback: Optional[str] = None

class Shakespeare(Workflow):

    llm = BedrockConverse(
        model="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    )

    @step
    async def generate_post(self, ctx: Context, ev: PostEvent) -> VerifyPostEvent:
        feedback = ev.feedback
        topic = ev.topic
        await ctx.store.set("topic", topic)
        retries = await ctx.store.get("retries", default=0)
        print(f"Generating Post using {topic=}, {feedback=}, {retries=}")
        system_prompt = """
Thou art a witty bard, renowned for turning the mundane into the
magnificent with thy playful jests and biting sarcasm. Armed with wit and
wisdom, thou dost revel in the creation of humorous quips most pleasing to the ear.
"""
        shakespeare_agent = ReActAgent(
            name="shakespeare_agent",
            system_prompt=system_prompt,
            llm=self.llm,
        )
        prompt = f"""
 Given the topic '{topic}', compose a humorous hot take in the style of Shakespeare. 
The tone should be sarcastic and playful. The final short form social media post 
must be over 200 characters and not exceed 280 characters, and emojis are strictly forbidden.

Please incorporate the following feedback if present: 
{feedback}

"""
        response = await shakespeare_agent.run(user_msg=prompt)
        await ctx.store.set("retries", retries + 1)
        return VerifyPostEvent(post=str(response))

    @step
    async def verify_post(
        self, ctx: Context, ev: VerifyPostEvent
    ) -> StopEvent | PostEvent:
        post = ev.post
        retries = await ctx.store.get("retries")
        if retries > 1:
            return StopEvent(result="Failed to generate a valid post after 3 retries.")
        system_prompt = """
You are a careful reviewer, skilled at understanding the core message of a post. 
Your job is to maintain the clarity and brevity of the post by ensuring it contains no emojis, 
unnecessary commentary, or excessive verbosity.
"""
        print(f"Verifying Post: {post=}")
        post_verifier_agent = ReActAgent(
            name="post_verifier_agent",
            system_prompt=system_prompt,
            llm=self.llm,
            output_cls=VerifyPostOutput,
        )
        prompt = f"""
Verify that the given X post meets the following criteria:
    - It is between 200 and 280 characters inclusive.
    - It contains no emojis.
    - It contains only the post itself, without additional commentary.

    The post should follow the 1-3-1 rule:
    - 1 bold statement to hook the reader
    - 3 lines of supporting information
    - 1 sentence to summarize the post

    Additionally, if you believe there are any issues with the post 
    or ways it could be improved, such as the structure of the post,
    rhythm, word choice, please provide feedback.

    If any of the criteria are not met, the post is considered invalid.
    Provide actionable changes about what is wrong and what actions
    need to be taken to fix the post.
    
    Your final response must include:
    - Valid: True/False
    - Feedback: Provide commentary if the post fails any of the criteria.

    X Post to Verify: 
    {post}
"""
        response = await post_verifier_agent.run(user_msg=prompt)

        print(f"Response: {response.structured_response}")
        topic = await ctx.store.get("topic")
        print(f"TOPIC: {topic}")
        if response.structured_response["valid"]:
            return StopEvent(result=str(response))
        else:
            return PostEvent(
                topic=topic, feedback=response.structured_response["feedback"]
            )

async def run():
    w = Shakespeare(timeout=60, verbose=False)
    ctx = Context(w)
    result = await w.run(topic="pirates", ctx=ctx)
    print(str(result))

if __name__ == "__main__":
    asyncio.run(run())

from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field
import asyncio

class GraphState(TypedDict):
    topic: str
    post: Optional[str]
    feedback: Optional[str]
    retries: int
    valid: bool
    final_result: Optional[str]

class VerifyPostOutput(BaseModel):
    valid: bool = Field(description="Whether the post meets all criteria")
    feedback: Optional[str] = Field(description="Feedback if the post is invalid")

llm = ChatBedrockConverse(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

@tool
def count_characters(text: str) -> int:
    """Count the number of characters in a text string."""
    return len(text)

@tool
def check_for_emojis(text: str) -> bool:
    """Check if the text contains any emojis."""
    return any(
        ord(char) > 127 and ord(char) not in range(0x2000, 0x206F) for char in text
    )

async def generate_post(state: GraphState) -> GraphState:
    """Generate a Shakespeare-style post based on the topic and feedback."""
    topic = state["topic"]
    feedback = state.get("feedback")
    retries = state["retries"]

    print(f"Generating Post using {topic=}, {feedback=}, {retries=}")

    shakespeare_agent = create_react_agent(
        llm,
        tools=[count_characters],
        state_modifier="""Thou art a witty bard, renowned for turning the mundane into the
magnificent with thy playful jests and biting sarcasm. Armed with wit and
wisdom, thou dost revel in the creation of humorous quips most pleasing to the ear.""",
    )

    user_prompt = f"""Given the topic '{topic}', compose a humorous hot take in the style of Shakespeare. 
The tone should be sarcastic and playful. The final short form social media post 
must be over 200 characters and not exceed 280 characters, and emojis are strictly forbidden.

Please incorporate the following feedback if present: 
{feedback if feedback else 'None'}

Use the count_characters tool to verify your post length.
Return ONLY the post text in your final response."""

    result = await shakespeare_agent.ainvoke(
        {"messages": [HumanMessage(content=user_prompt)]}
    )

    post = result["messages"][-1].content.strip()

    lines = post.split("\n")

    post = post.strip("\"'")

    return {**state, "post": post, "retries": retries + 1}

async def verify_post(state: GraphState) -> GraphState:
    """Verify that the generated post meets all criteria."""
    post = state["post"]
    retries = state["retries"]

    print(f"Verifying Post: {post=}")

    if retries > 3:
        return {
            **state,
            "valid": False,
            "final_result": "Failed to generate a valid post after 3 retries.",
        }

    verification_agent = create_react_agent(
        llm,
        tools=[count_characters, check_for_emojis],
        state_modifier="""You are a careful reviewer, skilled at understanding the core message of a post. 
Your job is to maintain the clarity and brevity of the post by ensuring it contains no emojis, 
unnecessary commentary, or excessive verbosity.""",
        response_format=VerifyPostOutput, # Make sure the model supports with_structured_output
    )

    user_prompt = f"""Verify that the given X post meets the following criteria:
    - It is between 200 and 280 characters inclusive.
    - It contains no emojis.
    - It contains only the post itself, without additional commentary.

    The post should follow the 1-3-1 rule:
    - 1 bold statement to hook the reader
    - 3 lines of supporting information
    - 1 sentence to summarize the post

    Use the available tools to check character count and emoji presence.

    If any criteria are not met, provide actionable feedback about what needs to be fixed.
    
    X Post to Verify: 
    {post}"""

    result = await verification_agent.ainvoke(
        {"messages": [HumanMessage(content=user_prompt)]}
    )

    verification_result = result["structured_response"]

    if verification_result.valid:
        return {**state, "valid": True, "final_result": post}
    else:
        return {**state, "valid": False, "feedback": verification_result.feedback}

def should_retry(state: GraphState) -> Literal["generate", "end"]:
    """Determine whether to retry generation or end the workflow."""
    if state["valid"] or state.get("final_result"):
        return "end"
    else:
        return "generate"

def create_shakespeare_workflow():
    """Create and compile the Shakespeare workflow graph."""
    workflow = StateGraph(GraphState)

    workflow.add_node("generate", generate_post)
    workflow.add_node("verify", verify_post)

    workflow.set_entry_point("generate")

    workflow.add_edge("generate", "verify")
    workflow.add_conditional_edges(
        "verify", should_retry, {"generate": "generate", "end": END}
    )

    return workflow.compile()

async def run():
    """Run the Shakespeare workflow with a given topic."""

    app = create_shakespeare_workflow()

    initial_state = {
        "topic": "pirates",
        "post": None,
        "feedback": None,
        "retries": 0,
        "valid": False,
        "final_result": None,
    }

    result = await app.ainvoke(initial_state)

    print("\n=== Final Result ===")
    if result["final_result"]:
        print(f"Post: {result['final_result']}")
        print(f"Character count: {len(result['final_result'])}")
        print(f"Attempts: {result['retries']}")
    else:
        print("Failed to generate a valid post")

if __name__ == "__main__":
    asyncio.run(run())