AI Summary: AI agents are transforming technology interaction by automating complex tasks through intelligent systems that can decide, learn, and collaborate. Building these agents in 2025 is facilitated by powerful open-source frameworks, which offer cost-effectiveness, customization, security, and community support. Key features to consider in these frameworks include memory management, tool integration, multi-agent orchestration, reasoning capabilities, ease of deployment, monitoring, and language model flexibility.
import os
from typing import TypedDict, List, Dict, Any
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain
import langgraph.graph as lgraph
from langchain_openai import ChatOpenAI
# Define the State
class GraphState(TypedDict):
"""
Represents the state of our graph.
Attributes:
keys: Dictionary where we can store arbitrary values relevant to
our graph.
"""
keys: Dict[str, Any]
# Configure LLM
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
your_llm = ChatOpenAI(temperature=0.7)
def generate_outline(state: GraphState):
"""Generates a blog post outline."""
topic = state["keys"]["topic"]
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert blog post outline generator. Given a topic, you create a detailed and well-structured outline."),
("human", "Please generate a detailed outline for a blog post on the topic: {topic}")
])
chain = prompt | your_llm
outline = chain.invoke({"topic": topic})
return {"keys": {"outline": outline.content}}
def write_content(state: GraphState):
"""Writes the blog post content based on the outline."""
outline = state["keys"]["outline"]
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert blog post writer. Given an outline, you write high-quality, engaging, and informative content."),
("human", "Please write a blog post based on the following outline:\\
{outline}")
])
chain = prompt | your_llm
content = chain.invoke({"outline": outline})
return {"keys": {"content": content.content}}
# Define the Graph
workflow = lgraph.GraphState(GraphState)
workflow.add_node("generate_outline", generate_outline)
workflow.add_node("write_content", write_content)
workflow.set_entry_point("generate_outline")
workflow.add_edge("generate_outline", "write_content")
# Compile the graph
app = workflow.compile()
# Run the Workflow
topic = "The Future of AI in Education"
state = {"keys": {"topic": topic}}
result = app.invoke(state)
print("Blog Post Outline:", result['keys']['outline'])
print("Blog Post Content:", result['keys']['content'])
# pip install -U autogen-agentchat autogen-ext[openai,web-surfer]
# playwright install
import asyncio
from autogen_agentchat.agents import UserProxyAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.agents.web_surfer import MultimodalWebSurfer
async def main() -> None:
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# The web surfer will open a Chromium browser window to perform web browsing tasks.
web_surfer = MultimodalWebSurfer("web_surfer", model_client, headless=False, animate_actions=True)
# The user proxy agent is used to get user input after each step of the web surfer.
# NOTE: you can skip input by pressing Enter.
user_proxy = UserProxyAgent("user_proxy")
# The termination condition is set to end the conversation when the user types 'exit'.
termination = TextMentionTermination("exit", sources=["user_proxy"])
# Web surfer and user proxy take turns in a round-robin fashion.
team = RoundRobinGroupChat([web_surfer, user_proxy], termination_condition=termination)
try:
# Start the team and wait for it to terminate.
await Console(team.run_stream(task="Find information about AutoGen and write a short summary."))
finally:
await web_surfer.close()
await model_client.close()
asyncio.run(main())
from typing import List
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
# Check our tools documentations for more information on how to use them
from crewai_tools import SerperDevTool, ScrapeWebsiteTool, WebsiteSearchTool, FileReadTool
from pydantic import BaseModel, Field
web_search_tool = WebsiteSearchTool()
seper_dev_tool = SerperDevTool()
file_read_tool = FileReadTool(
file_path='job_description_example.md',
description='A tool to read the job description example file.'
)
class ResearchRoleRequirements(BaseModel):
"""Research role requirements model"""
skills: List[str] = Field(..., description="List of recommended skills for the ideal candidate aligned with the company's culture, ongoing projects, and the specific role's requirements.")
experience: List[str] = Field(..., description="List of recommended experience for the ideal candidate aligned with the company's culture, ongoing projects, and the specific role's requirements.")
qualities: List[str] = Field(..., description="List of recommended qualities for the ideal candidate aligned with the company's culture, ongoing projects, and the specific role's requirements.")
@CrewBase
class JobPostingCrew:
"""JobPosting crew"""
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
@agent
def research_agent(self) -> Agent:
return Agent(
config=self.agents_config['research_agent'],
tools=[web_search_tool, seper_dev_tool],
verbose=True
)
@agent
def writer_agent(self) -> Agent:
return Agent(
config=self.agents_config['writer_agent'],
tools=[web_search_tool, seper_dev_tool, file_read_tool],
verbose=True
)
@agent
def review_agent(self) -> Agent:
return Agent(
config=self.agents_config['review_agent'],
tools=[web_search_tool, seper_dev_tool, file_read_tool],
verbose=True
)
@task
def research_company_culture_task(self) -> Task:
return Task(
config=self.tasks_config['research_company_culture_task'],
agent=self.research_agent()
)
@task
def research_role_requirements_task(self) -> Task:
return Task(
config=self.tasks_config['research_role_requirements_task'],
agent=self.research_agent(),
output_json=ResearchRoleRequirements
)
@task
def draft_job_posting_task(self) -> Task:
return Task(
config=self.tasks_config['draft_job_posting_task'],
agent=self.writer_agent()
)
@task
def review_and_edit_job_posting_task(self) -> Task:
return Task(
config=self.tasks_config['review_and_edit_job_posting_task'],
agent=self.review_agent()
)
@task
def industry_analysis_task(self) -> Task:
return Task(
config=self.tasks_config['industry_analysis_task'],
agent=self.research_agent()
)
@crew
def crew(self) -> Crew:
"""Creates the JobPostingCrew"""
return Crew(
agents=self.agents, # Automatically created by the @agent decorator
tasks=self.tasks, # Automatically created by the @task decorator
process=Process.sequential,
verbose=2,
)
from __future__ import annotations as _annotations
import asyncio
import random
import uuid
from pydantic import BaseModel
from agents import (
Agent,
HandoffOutputItem,
ItemHelpers,
MessageOutputItem,
RunContextWrapper,
Runner,
ToolCallItem,
ToolCallOutputItem,
TResponseInputItem,
function_tool,
handoff,
trace,
)
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
### CONTEXT
class AirlineAgentContext(BaseModel):
passenger_name: str | None = None
confirmation_number: str | None = None
seat_number: str | None = None
flight_number: str | None = None
### TOOLS
@function_tool(
name_override="faq_lookup_tool", description_override="Lookup frequently asked questions."
)
async def faq_lookup_tool(question: str) -> str:
if "bag" in question or "baggage" in question:
return (
"You are allowed to bring one bag on the plane. "
"It must be under 50 pounds and 22 inches x 14 inches x 9 inches."
)
elif "seats" in question or "plane" in question:
return (
"There are 120 seats on the plane. "
"There are 22 business class seats and 98 economy seats. "
"Exit rows are rows 4 and 16. "
"Rows 5-8 are Economy Plus, with extra legroom. "
)
elif "wifi" in question:
return "We have free wifi on the plane, join Airline-Wifi"
return "I'm sorry, I don't know the answer to that question."
@function_tool
async def update_seat(
context: RunContextWrapper[AirlineAgentContext], confirmation_number: str, new_seat: str
) -> str:
"""
Update the seat for a given confirmation number.
Args:
confirmation_number: The confirmation number for the flight.
new_seat: The new seat to update to.
"""
# Update the context based on the customer's input
context.context.confirmation_number = confirmation_number
context.context.seat_number = new_seat
# Ensure that the flight number has been set by the incoming handoff
assert context.context.flight_number is not None, "Flight number is required"
return f"Updated seat to {new_seat} for confirmation number {confirmation_number}"
### HOOKS
async def on_seat_booking_handoff(context: RunContextWrapper[AirlineAgentContext]) -> None:
flight_number = f"FLT-{random.randint(100, 999)}"
context.context.flight_number = flight_number
### AGENTS
faq_agent = Agent[AirlineAgentContext](
name="FAQ Agent",
handoff_description="A helpful agent that can answer questions about the airline.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are an FAQ agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
Use the following routine to support the customer.
# Routine
1. Identify the last question asked by the customer.
2. Use the faq lookup tool to answer the question. Do not rely on your own knowledge.
3. If you cannot answer the question, transfer back to the triage agent.""",
tools=[faq_lookup_tool],
)
seat_booking_agent = Agent[AirlineAgentContext](
name="Seat Booking Agent",
handoff_description="A helpful agent that can update a seat on a flight.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a seat booking agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
Use the following routine to support the customer.
# Routine
1. Ask for their confirmation number.
2. Ask the customer what their desired seat number is.
3. Use the update seat tool to update the seat on the flight.
If the customer asks a question that is not related to the routine, transfer back to the triage agent. """,
tools=[update_seat],
)
triage_agent = Agent[AirlineAgentContext](
name="Triage Agent",
handoff_description="A triage agent that can delegate a customer's request to the appropriate agent.",
instructions=(
f"{RECOMMENDED_PROMPT_PREFIX} "
"You are a helpful triaging agent. You can use your tools to delegate questions to other appropriate agents."
),
handoffs=[
faq_agent,
handoff(agent=seat_booking_agent, on_handoff=on_seat_booking_handoff),
],
)
faq_agent.handoffs.append(triage_agent)
seat_booking_agent.handoffs.append(triage_agent)
### RUN
async def main():
current_agent: Agent[AirlineAgentContext] = triage_agent
input_items: list[TResponseInputItem] = []
context = AirlineAgentContext()
# Normally, each input from the user would be an API request to your app, and you can wrap the request in a trace()
# Here, we'll just use a random UUID for the conversation ID
conversation_id = uuid.uuid4().hex[:16]
while True:
user_input = input("Enter your message: ")
with trace("Customer service", group_id=conversation_id):
input_items.append({"content": user_input, "role": "user"})
result = await Runner.run(current_agent, input_items, context=context)
for new_item in result.new_items:
agent_name = new_item.agent.name
if isinstance(new_item, MessageOutputItem):
print(f"{agent_name}: {ItemHelpers.text_message_output(new_item)}")
elif isinstance(new_item, HandoffOutputItem):
print(
f"Handed off from {new_item.source_agent.name} to {new_item.target_agent.name}"
)
elif isinstance(new_item, ToolCallItem):
print(f"{agent_name}: Calling a tool")
elif isinstance(new_item, ToolCallOutputItem):
print(f"{agent_name}: Tool call output: {new_item.output}")
else:
print(f"{agent_name}: Skipping item: {new_item.__class__.__name__}")
input_items = result.to_input_list()
current_agent = result.last_agent
if __name__ == "__main__":
asyncio.run(main())
"""
Example prompts to try:
- "What's the latest news and financial performance of Apple (AAPL)?"
- "Give me a detailed analysis of Tesla's (TSLA) current market position"
- "How are Microsoft's (MSFT) financials looking? Include analyst recommendations"
- "Analyze NVIDIA's (NVDA) stock performance and future outlook"
- "What's the market saying about Amazon's (AMZN) latest quarter?"
Run: `pip install openai yfinance agno` to install the dependencies
"""
from textwrap import dedent
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.yfinance import YFinanceTools
finance_agent = Agent(
model=OpenAIChat(id="gpt-4o"),
tools=[
YFinanceTools(
stock_price=True,
analyst_recommendations=True,
stock_fundamentals=True,
historical_prices=True,
company_info=True,
company_news=True,
)
],
instructions=dedent("""\\
You are a seasoned Wall Street analyst with deep expertise in market analysis! 📊
Follow these steps for comprehensive financial analysis:
1. Market Overview
- Latest stock price
- 52-week high and low
2. Financial Deep Dive
- Key metrics (P/E, Market Cap, EPS)
3. Professional Insights
- Analyst recommendations breakdown
- Recent rating changes
4. Market Context
- Industry trends and positioning
- Competitive analysis
- Market sentiment indicators
Your reporting style:
- Begin with an executive summary
- Use tables for data presentation
- Include clear section headers
- Add emoji indicators for trends (📈 📉)
- Highlight key insights with bullet points
- Compare metrics to industry averages
- Include technical term explanations
- End with a forward-looking analysis
Risk Disclosure:
- Always highlight potential risk factors
- Note market uncertainties
- Mention relevant regulatory concerns
"""),
add_datetime_to_instructions=True,
show_tool_calls=True,
markdown=True,
)
# Example usage with detailed market analysis request
finance_agent.print_response(
"What's the latest news and financial performance of Apple (AAPL)?", stream=True
)
# Semiconductor market analysis example
finance_agent.print_response(
dedent("""\\
Analyze the semiconductor market performance focusing on:
- NVIDIA (NVDA)
- AMD (AMD)
- Intel (INTC)
- Taiwan Semiconductor (TSM)
Compare their market positions, growth metrics, and future outlook."""),
stream=True,
)
# Automotive market analysis example
finance_agent.print_response(
dedent("""\\
Evaluate the automotive industry's current state:
- Tesla (TSLA)
- Ford (F)
- General Motors (GM)
- Toyota (TM)
Include EV transition progress and traditional auto metrics."""),
stream=True,
)
# More example prompts to explore:
"""
Advanced analysis queries:
1. "Compare Tesla's valuation metrics with traditional automakers"
2. "Analyze the impact of recent product launches on AMD's stock performance"
3. "How do Meta's financial metrics compare to its social media peers?"
4. "Evaluate Netflix's subscriber growth impact on financial metrics"
5. "Break down Amazon's revenue streams and segment performance"
Industry-specific analyses:
Semiconductor Market:
1. "How is the chip shortage affecting TSMC's market position?"
2. "Compare NVIDIA's AI chip revenue growth with competitors"
3. "Analyze Intel's foundry strategy impact on stock performance"
4. "Evaluate semiconductor equipment makers like ASML and Applied Materials"
Automotive Industry:
1. "Compare EV manufacturers' production metrics and margins"
2. "Analyze traditional automakers' EV transition progress"
3. "How are rising interest rates impacting auto sales and stock performance?"
4. "Compare Tesla's profitability metrics with traditional auto manufacturers"
"""
import os
from metagpt.roles.di.data_interpreter import DataInterpreter
async def main():
email_account = "your_email_account"
# your password will stay only on your device and not go to LLM api
os.environ["email_password"] = "your_email_password"
### Prompt for automatic email reply, uncomment to try this too ###
# prompt = f"""I will give you your Outlook email account ({email_account}) and password (email_password item in the environment variable). You need to find the latest email in my inbox with the sender's suffix @gmail.com and reply "Thank you! I have received your email~"""""
### Prompt for automatic email summary ###
prompt = f"""I will give you your Outlook email account ({email_account}) and password (email_password item in the environment variable).
Firstly, Please help me fetch the latest 5 senders and full letter contents.
Then, summarize each of the 5 emails into one sentence (you can do this by yourself, no need to import other models to do this) and output them in a markdown format."""
di = DataInterpreter()
await di.run(prompt)
if __name__ == "__main__":
import asyncio
asyncio.run(main())