#Practical Examples

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search = SerperDevTool()

researcher = Agent(
    role="Senior Researcher",
    goal="Gather accurate, dated facts about {topic}",
    backstory="You verify claims and prefer primary sources.",
    tools=[search],
    verbose=True,
)

writer = Agent(
    role="Technical Writer",
    goal="Turn research notes into a briefing engineers will finish",
    backstory="You write short paragraphs and never invent citations.",
    verbose=True,
)

research_task = Task(
    description="Research {topic}. Use web search. Note dates and sources.",
    expected_output="Markdown bullets: fact, source, why it matters.",
    agent=researcher,
)

write_task = Task(
    description="Write a 400-word briefing from the research notes.",
    expected_output="Markdown with Summary, Findings, Risks. No wrapping fence.",
    agent=writer,
    context=[research_task],
    markdown=True,
    output_file="output/briefing.md",
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
    memory=True,
)

result = crew.kickoff(inputs={"topic": "open-weight LLM serving"})
print(result.raw)

from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, listen, start

class PipelineState(BaseModel):
    topic: str = ""
    briefing: str = ""

def make_crew() -> Crew:
    agent = Agent(
        role="Briefing researcher",
        goal="Produce a sourced outline of {topic}",
        backstory="You are skeptical of marketing copy.",
    )
    task = Task(
        description="Outline {topic} in 8 bullets with one risk each.",
        expected_output="Eight markdown bullets.",
        agent=agent,
    )
    return Crew(agents=[agent], tasks=[task], process=Process.sequential)

class BriefingPipeline(Flow[PipelineState]):
    @start()
    def set_topic(self):
        self.state.topic = "CrewAI vs LangGraph"

    @listen(set_topic)
    def research(self):
        out = make_crew().kickoff(inputs={"topic": self.state.topic})
        self.state.briefing = out.raw
        return out.raw

    @listen(research)
    def done(self):
        print("chars:", len(self.state.briefing))
        return self.state.briefing

if __name__ == "__main__":
    print(BriefingPipeline().kickoff())

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

FAQ = {
    "flows": "Flows own state and control; use @start and @listen.",
    "crews": "Crews are role-playing teams; Process is sequential or hierarchical.",
}

class SearchInput(BaseModel):
    query: str = Field(..., description="Keyword: flows or crews")

class FaqSearchTool(BaseTool):
    name: str = "faq_search"
    description: str = "Search the internal FAQ. Use for CrewAI concept questions."
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        q = query.lower()
        hits = [v for k, v in FAQ.items() if k in q or q in k]
        return "\n".join(hits) if hits else "No FAQ hit. Say you do not know."

agent = Agent(
    role="Academy tutor",
    goal="Answer only from faq_search",
    backstory="You refuse to guess when the tool is empty.",
    tools=[FaqSearchTool()],
)

task = Task(
    description="Explain how Flows relate to Crews for a new student.",
    expected_output="Four sentences max, grounded in tool output.",
    agent=agent,
)

print(Crew(agents=[agent], tasks=[task], process=Process.sequential).kickoff().raw)