상태 관리
상태 관리 (State & Context)
지금까지 멀티에이전트 애플리케이션의 컴포넌트 — 에이전트, 팀, 종료 조건 — 를 만드는 방법을 살펴봤어요. 많은 경우 이 컴포넌트들의 상태를 디스크에 저장하고 나중에 다시 로드하는 것이 유용해요. 특히 무상태(stateless) 엔드포인트가 요청에 응답하고 영구 저장소에서 애플리케이션의 상태를 로드해야 하는 웹 애플리케이션에서 유용하죠.
이 노트북에서는 에이전트, 팀, 종료 조건의 상태를 저장하고 로드하는 방법을 다룰게요.
에이전트 저장·로드
AssistantAgent의 save_state 메서드를 호출하면 에이전트의 상태를 얻을 수 있어요.
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")
assistant_agent = AssistantAgent(
name="assistant_agent",
system_message="You are a helpful assistant",
model_client=model_client,
)
# Use asyncio.run(...) when running in a script.
response = await assistant_agent.on_messages(
[TextMessage(content="Write a 3 line poem on lake tangayika", source="user")], CancellationToken()
)
print(response.chat_message)
await model_client.close()
agent_state = await assistant_agent.save_state()
print(agent_state)
model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")
new_assistant_agent = AssistantAgent(
name="assistant_agent",
system_message="You are a helpful assistant",
model_client=model_client,
)
await new_assistant_agent.load_state(agent_state)
# Use asyncio.run(...) when running in a script.
response = await new_assistant_agent.on_messages(
[TextMessage(content="What was the last line of the previous poem you wrote", source="user")], CancellationToken()
)
print(response.chat_message)
await model_client.close()
AssistantAgent의 상태는 model_context로 구성돼요. 직접 커스텀 에이전트를 작성한다면,BaseChatAgent.save_state와BaseChatAgent.load_state메서드를 오버라이드해 동작을 맞춤 설정하는 것을 고려하세요. 기본 구현은 빈 상태를 저장·로드해요.
팀 저장·로드
팀에서 save_state 메서드를 호출해 팀의 상태를 얻고, load_state 메서드로 로드할 수 있어요.
팀에서 save_state를 호출하면 팀 안의 모든 에이전트의 상태를 저장해요.
단일 에이전트가 있는 간단한 RoundRobinGroupChat 팀을 만들고 시(poem)를 쓰게 해 볼게요.
model_client = OpenAIChatCompletionClient(model="gpt-4o-2024-08-06")
# Define a team.
assistant_agent = AssistantAgent(
name="assistant_agent",
system_message="You are a helpful assistant",
model_client=model_client,
)
agent_team = RoundRobinGroupChat([assistant_agent], termination_condition=MaxMessageTermination(max_messages=2))
# Run the team and stream messages to the console.
stream = agent_team.run_stream(task="Write a beautiful poem 3-line about lake tangayika")
# Use asyncio.run(...) when running in a script.
await Console(stream)
# Save the state of the agent team.
team_state = await agent_team.save_state()
팀을 리셋하면(팀 인스턴스화를 시뮬레이션), What was the last line of the poem you wrote?라는 질문에 팀이 답하지 못하는 걸 볼 수 있어요. 이전 실행에 대한 참조가 없기 때문이죠.
await agent_team.reset()
stream = agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)
다음으로 팀의 상태를 로드하고 같은 질문을 해 볼게요. 팀이 자신이 쓴 시의 마지막 줄을 정확히 반환하는 걸 볼 수 있어요.
print(team_state)
# Load team state.
await agent_team.load_state(team_state)
stream = agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)
상태 영속화 (파일 또는 데이터베이스)
많은 경우 팀의 상태를 디스크(또는 데이터베이스)에 **영속화(persist)**하고 나중에 다시 로드하고 싶을 거예요. 상태는 파일에 직렬화하거나 데이터베이스에 쓸 수 있는 딕셔너리예요.
import json
## save state to disk
with open("coding/team_state.json", "w") as f:
json.dump(team_state, f)
## load state from disk
with open("coding/team_state.json", "r") as f:
team_state = json.load(f)
new_agent_team = RoundRobinGroupChat([assistant_agent], termination_condition=MaxMessageTermination(max_messages=2))
await new_agent_team.load_state(team_state)
stream = new_agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)
await model_client.close()