Agno 프레임워크로 AI 에이전트 구축하기

# 가상환경 생성 (선택사항이지만 권장)
python -m venv agno_env
source agno_env/bin/activate  # Linux/Mac
# 또는 Windows에서는: agno_env\Scripts\activate

# Agno 설치
pip install agno

# basic_agent.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat  # OpenAI 모델 사용

# API 키 설정 (또는 환경변수로 설정)
import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key"

# 기본 에이전트 생성
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),  # OpenAI의 GPT-4o 모델 사용
    description="당신은 친절하고 도움이 되는 AI 비서입니다.",
    markdown=True  # 마크다운 형식 응답 활성화
)

# 에이전트에게 질문하고 응답 출력
agent.print_response("오늘 서울 날씨가 어때?", stream=True)

# agent_with_tools.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools  # 웹 검색 도구

# API 키 설정
import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key"

# 도구를 가진 에이전트 생성
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    description="당신은 최신 정보를 제공하는 AI 비서입니다.",
    tools=[DuckDuckGoTools()],  # DuckDuckGo 검색 도구 추가
    show_tool_calls=True,  # 도구 호출 과정 표시
    markdown=True
)

# 에이전트에게 질문하고 응답 출력
agent.print_response("오늘 서울의 날씨는 어때?", stream=True)

# knowledge_agent.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.lancedb import LanceDb, SearchType

# API 키 설정
import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key"

# 지식 에이전트 생성
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    description="당신은 한식 요리 전문가입니다.",
    instructions=[
        "지식 베이스에서 한국 요리 레시피를 검색하세요.",
        "질문이 지식 베이스의 내용에 관련되지 않으면 솔직히 모른다고 답변하세요."
    ],
    knowledge=PDFUrlKnowledgeBase(
        urls=["https://example.com/korean_recipes.pdf"],  # 한식 레시피 PDF URL
        vector_db=LanceDb(
            uri="tmp/lancedb",
            table_name="recipes",
            search_type=SearchType.hybrid,
            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        ),
    ),
    markdown=True
)

# 지식 베이스 로드 (처음 한 번만 실행)
if agent.knowledge is not None:
    agent.knowledge.load()

# 에이전트에게 질문
agent.print_response("김치찌개를 어떻게 만들어?", stream=True)

# mcp_agent.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools

# MCP 파일시스템 서버와 연결하는 에이전트 생성
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[
        MCPTools(
            command="npx @mcp/server-filesystem@latest",
            tool_names=["filesystem"],
            add_instructions=True,
        )
    ],
    show_tool_calls=True,
    markdown=True
)

# 에이전트에게 파일시스템 관련 작업 요청
agent.print_response("현재 디렉토리에 있는 파일들을 보여주고, 가장 큰 파일의 내용을 출력해줘", stream=True)

# agent_team.py
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools
from agno.team import Team

# 웹 검색 전문 에이전트
web_agent = Agent(
    name="웹 검색 에이전트",
    role="웹에서 정보를 검색합니다",
    model=OpenAIChat(id="gpt-4o"),
    tools=[DuckDuckGoTools()],
    instructions="항상 출처를 포함해주세요",
    markdown=True
)

# 금융 정보 전문 에이전트
finance_agent = Agent(
    name="금융 에이전트",
    role="금융 데이터를 분석합니다",
    model=OpenAIChat(id="gpt-4o"),
    tools=[YFinanceTools(stock_price=True, company_info=True)],
    instructions="데이터를 표로 정리해 보여주세요",
    markdown=True
)

# 에이전트 팀 구성
agent_team = Team(
    mode="coordinate",  # 협력 모드 설정 (route, collaborate, coordinate 중 선택)
    members=[web_agent, finance_agent],  # 팀 구성원
    model=OpenAIChat(id="gpt-4o"),  # 팀 조정자 모델
    success_criteria="데이터에 기반한 통찰력 있는 종합 보고서",
    instructions=["항상 출처 포함", "데이터는 표로 정리"],
    show_tool_calls=True,
    markdown=True
)

# 팀에게 복합적인 질문
agent_team.print_response("AI 반도체 기업들의 시장 전망과 재무 성과는 어떤가요?", stream=True)

# app.py
import os
import streamlit as st
import time
from typing import Iterator

# Agno 라이브러리 임포트
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.lancedb import LanceDb, SearchType
from agno.run.response import RunResponse

# 페이지 설정
st.set_page_config(
    page_title="한식 레시피 도우미",
    page_icon="🍲",
    layout="wide"
)

# 사이드바 - API 키 설정
with st.sidebar:
    st.title("🔑 API 키 설정")
    openai_api_key = st.text_input("OpenAI API 키", type="password")
    
    st.title("ℹ️ 정보")
    st.info("""
    이 애플리케이션은 Agno 라이브러리를 사용한 지식 에이전트입니다.
    한국 요리 레시피에 관한 질문에 답변해 드립니다.
    """)
    
    # 디버그 모드 체크박스
    debug_mode = st.checkbox("디버그 모드", value=False)
    
    # 지식 베이스 재로드 버튼
    if st.button("지식 베이스 재로드"):
        st.session_state.knowledge_loaded = False
        st.rerun()

# 메인 영역 - 제목 및 설명
st.title("🍲 한식 레시피 도우미")
st.markdown("""
한국 요리 레시피와 관련된 질문을 해보세요. 예시:
- 김치찌개 레시피가 궁금해요
- 비빔밥에 들어가는 재료는 무엇인가요?
- 불고기 양념을 만드는 방법을 알려주세요
- 떡볶이와 라볶이의 차이점은 무엇인가요?
""")

# 전역 변수로 Agent 초기화 함수
def initialize_agent():
    """지식 에이전트를 초기화합니다."""
    # API 키 설정
    os.environ["OPENAI_API_KEY"] = openai_api_key
    
    # PDF URL (한국 요리 레시피)
    recipe_pdf_url = "https://cdn-akamai.lkk.com/-/media/kr-site---homecook/e-book_kr/korean-cuisine-recipe-book-2021_kr.pdf"
    
    # 지식 에이전트 생성
    agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        description="당신은 한식 요리 전문가입니다.",
        instructions=[
            "지식 베이스에서 한국 요리 레시피를 검색하세요.",
            "레시피 외에도 한국 음식 문화와 역사에 대한 정보를 제공하세요.",
            "답변을 찾을 수 없다면 솔직히 모른다고 답변하세요.",
            "식재료와 조리 방법을 자세히 설명하세요.",
            "모든 답변에 출처를 포함하세요."
        ],
        knowledge=PDFUrlKnowledgeBase(
            urls=[recipe_pdf_url],
            vector_db=LanceDb(
                uri="tmp/lancedb",
                table_name="korean_recipes",
                search_type=SearchType.hybrid,
                embedder=OpenAIEmbedder(id="text-embedding-3-small"),
            ),
        ),
        markdown=True,
        debug_mode=debug_mode
    )
    
    return agent

# 세션 상태 초기화
if "messages" not in st.session_state:
    st.session_state.messages = []

if "knowledge_loaded" not in st.session_state:
    st.session_state.knowledge_loaded = False

# 이전 메시지 표시
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 사용자 입력 처리
user_query = st.chat_input("질문을 입력하세요...")

if user_query:
    # 사용자 입력을 대화 기록에 추가
    st.session_state.messages.append({"role": "user", "content": user_query})
    
    # 사용자 메시지 표시
    with st.chat_message("user"):
        st.markdown(user_query)
    
    # 로딩 상태 표시
    with st.chat_message("assistant"):
        if not openai_api_key:
            st.error("OpenAI API 키를 입력해주세요.")
            st.session_state.messages.append({"role": "assistant", "content": "⚠️ OpenAI API 키를 입력해주세요."})
        else:
            message_placeholder = st.empty()
            full_response = ""
            
            try:
                # 에이전트 초기화
                agent = initialize_agent()
                
                # 지식 베이스 로드
                if not st.session_state.knowledge_loaded:
                    with st.spinner("지식 베이스를 로딩 중입니다..."):
                        if agent.knowledge is not None:
                            agent.knowledge.load()
                            st.session_state.knowledge_loaded = True
                
                # 에이전트 스트리밍 응답
                response_stream: Iterator[RunResponse] = agent.run(user_query, stream=True)
                
                for chunk in response_stream:
                    if chunk.content and isinstance(chunk.content, str):
                        full_response += chunk.content
                        message_placeholder.markdown(full_response + "▌")
                
                message_placeholder.markdown(full_response)
                
                # 응답을 대화 기록에 추가
                st.session_state.messages.append({"role": "assistant", "content": full_response})
            
            except Exception as e:
                error_message = f"오류가 발생했습니다: {str(e)}"
                st.error(error_message)
                message_placeholder.markdown(f"⚠️ {error_message}")
                st.session_state.messages.append({"role": "assistant", "content": f"오류: {str(e)}"})

async def run_agent(message: str):
    """여행 계획 에이전트 팀을 실행합니다."""
    # API 키 가져오기
    google_maps_key = st.session_state.get('google_maps_key')
    accuweather_key = st.session_state.get('accuweather_key')
    openai_key = st.session_state.get('openai_key')
    
    # 환경 설정
    env = {
        **os.environ,
        "GOOGLE_MAPS_API_KEY": google_maps_key,
        "ACCUWEATHER_API_KEY": accuweather_key,
        "OPENAI_API_KEY": openai_key
    }

    # MCP 도구 설정
    async with MultiMCPTools(
        [
            "npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",  # Airbnb MCP
            "npx -y @modelcontextprotocol/server-google-maps",  # Google Maps MCP
            "uvx --from git+https://github.com/adhikasp/mcp-weather.git mcp-weather",  # 날씨 MCP
            "./calendar_mcp.py"  # 커스텀 캘린더 MCP
        ],
        env=env,
    ) as mcp_tools:
        # 전문 에이전트 정의
        maps_agent = Agent(
            tools=[mcp_tools],
            model=OpenAIChat(id="gpt-4o-mini", api_key=openai_key),
            name="Maps Agent",
            goal="최적 경로 찾기, 관심 지점 식별, 이동 시간 계산 등"
        )
        
        weather_agent = Agent(
            tools=[mcp_tools],
            name="Weather Agent",
            model=OpenAIChat(id="gpt-4o-mini", api_key=openai_key),
            goal="목적지 날씨 예보 제공, 기상 조건에 따른 활동 추천 등"
        )
        
        booking_agent = Agent(
            tools=[mcp_tools],
            name="Booking Agent",
            model=OpenAIChat(id="gpt-4o-mini", api_key=openai_key),
            goal="예산 내 숙박 시설 찾기, 가격 비교, 특정 날짜 가용성 확인 등"
        )
        
        calendar_agent = Agent(
            tools=[mcp_tools],
            name="Calendar Agent",
            model=OpenAIChat(id="gpt-4o-mini", api_key=openai_key),
            goal="상세 여행 일정 작성, 예약 및 체크인 알림 설정, 활동 일정 관리 등"
        )

        # 에이전트 팀 구성
        team = Team(
            members=[maps_agent, weather_agent, booking_agent, calendar_agent],
            name="Travel Planning Team",
            markdown=True,
            show_tool_calls=True,
            instructions="종합적인 여행 계획 작성을 위해 협력하세요."
        )
        
        result = await team.arun(message)
        output = result.messages[-1].content
        return output