AgentScope 시작하기 — 설치부터 첫 에이전트까지

pip install agentscope

pip install agentscope[models]

pip install agentscope[full]

pip install agentscope[rag]        # RAG + 벡터 저장소
pip install agentscope[memory]     # Redis + Mem0 + ReMe 메모리
pip install agentscope[a2a]        # Agent-to-Agent 프로토콜
pip install agentscope[realtime]   # 실시간 음성
pip install agentscope[tuner]      # RL 파인튜닝

from agentscope.agent import ReActAgent

agent = ReActAgent(
    name="Friday",
    sys_prompt="You are a helpful coding assistant.",
    model=model,
    memory=memory,
    toolkit=toolkit,
)

from agentscope.model import OpenAIChatModel

model = OpenAIChatModel(
    model_name="gpt-4o",
    api_key="sk-...",
    stream=True,
)

from agentscope.memory import InMemoryMemory

memory = InMemoryMemory()

from agentscope.tool import Toolkit, execute_python_code, execute_shell_command

toolkit = Toolkit()
toolkit.register_tool_function(execute_python_code)
toolkit.register_tool_function(execute_shell_command)

def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is sunny, 22°C"

toolkit.register_tool_function(get_weather)

from agentscope.formatter import OpenAIChatFormatter

formatter = OpenAIChatFormatter()

from agentscope.agent import ReActAgent, UserAgent
from agentscope.model import OpenAIChatModel
from agentscope.formatter import OpenAIChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.tool import Toolkit, execute_python_code
import asyncio, os

async def main():
    # 1. 도구 등록
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    # 2. 에이전트 생성
    agent = ReActAgent(
        name="DataAnalyst",
        sys_prompt="You are a data analyst. Use Python to analyze data and answer questions.",
        model=OpenAIChatModel(
            model_name="gpt-4o",
            api_key=os.environ["OPENAI_API_KEY"],
            stream=True,
        ),
        memory=InMemoryMemory(),
        formatter=OpenAIChatFormatter(),
        toolkit=toolkit,
    )

    # 3. 유저 에이전트 (터미널 입력)
    user = UserAgent(name="user")

    # 4. 대화 루프
    msg = None
    while True:
        msg = await agent(msg)
        msg = await user(msg)
        if msg.get_text_content().lower() == "exit":
            break

asyncio.run(main())

user: Calculate the first 20 Fibonacci numbers
DataAnalyst: [Thinking] I need to write Python code to generate Fibonacci numbers.
DataAnalyst: [Tool Call] execute_python_code
DataAnalyst: The first 20 Fibonacci numbers are: 0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987, 1597, 2584, 4181

import httpx

def search_web(query: str, max_results: int = 3) -> str:
    """Search the web using DuckDuckGo and return top results.

    Args:
        query: The search query string.
        max_results: Maximum number of results to return.

    Returns:
        A formatted string of search results.
    """
    url = "https://api.duckduckgo.com/"
    params = {"q": query, "format": "json", "no_html": 1}
    resp = httpx.get(url, params=params, timeout=10)
    data = resp.json()

    results = []
    for item in data.get("RelatedTopics", [])[:max_results]:
        if "Text" in item:
            results.append(item["Text"])

    return "\n".join(results) if results else "No results found."

toolkit.register_tool_function(search_web)

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float

agent = ReActAgent(
    name="Analyst",
    sys_prompt="Analyze the given data and return structured results.",
    model=model,
    memory=memory,
    formatter=formatter,
    toolkit=toolkit,
    structured_output=AnalysisResult,  # Pydantic 모델 지정
)

from agentscope.agent import CompressionConfig

agent = ReActAgent(
    name="Friday",
    sys_prompt="...",
    model=model,
    memory=memory,
    formatter=formatter,
    toolkit=toolkit,
    compression=CompressionConfig(
        max_tokens=4000,       # 이 토큰 수를 초과하면 압축
        keep_recent=5,         # 최근 5개 메시지는 유지
    ),
)

async def logging_hook(agent, msg):
    print(f"[LOG] {agent.name} received: {msg.get_text_content()[:50]}")
    return msg

agent = ReActAgent(
    name="Friday",
    sys_prompt="...",
    model=model,
    memory=memory,
    formatter=formatter,
    toolkit=toolkit,
    hooks={
        "pre_reply": [logging_hook],
        "post_reply": [logging_hook],
    },
)