> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreweave.com/llms.txt
> Use this file to discover all available pages before exploring further.
# AutoGen
> Weave を使用して、AutoGen のエージェントやマルチエージェントシステムをトラッキングおよび監視します
AutoGen は、AI エージェントやアプリケーションを構築するための Microsoft 製フレームワークで、複雑なマルチエージェントシステムを簡単に作成できます。AutoGen には、会話型 AI 向けのコンポーネント (`AgentChat`) 、マルチエージェントのコア機能 (`Core`) 、外部サービスとのインテグレーション (`Extensions`) が用意されています。また、ノーコードでエージェントのプロトタイプを作成できる `Studio` も提供されています。詳細については、[AutoGen の公式ドキュメント](https://microsoft.github.io/autogen/stable//index.html)を参照してください。
このガイドは、[AutoGen](https://microsoft.github.io/autogen/stable/index.html) の基本を理解していることを前提としています。
Weave は [AutoGen](https://microsoft.github.io/autogen/stable/index.html) と連携し、マルチエージェントアプリケーションの実行をトレースして可視化できます。Weave を初期化すると、`autogen_agentchat`、`autogen_core`、`autogen_ext` 内のやり取りが自動的にトラッキングされます。このガイドでは、AutoGen で Weave を設定する方法を説明するとともに、モデルクライアント、ツールを使用するエージェント、グループチャット、メモリ、RAG ワークフロー、エージェントランタイム、逐次ワークフロー、コード実行器を扱う実践的なサンプルを紹介します。このガイドを最後まで進めると、AutoGen アプリケーションの詳細なトレースを Weave で取得し、エージェントの動作をデバッグしたり、LLM の使用状況を監視したり、複雑なワークフローにおけるエージェント間のやり取りを把握したりできるようになります。
前提条件
始める前に、AutoGen と Weave をインストールしておく必要があります。また、使用する LLM プロバイダの SDK (OpenAI や Anthropic など) もインストールしておいてください。
```bash theme={"system"}
pip install autogen_agentchat "autogen_ext[openai,anthropic]" weave
```
以降のサンプルで使用する LLM プロバイダに対してモデルクライアントが認証できるよう、APIキーを環境変数として設定します。
```python lines theme={"system"}
import os
os.environ["OPENAI_API_KEY"] = "[YOUR-OPENAI-API-KEY]"
os.environ["ANTHROPIC_API_KEY"] = "[YOUR-ANTHROPIC-API-KEY]"
```
基本的なセットアップ
トレースの取得を開始するには、スクリプトの冒頭で Weave を初期化します。Weave を初期化すると、それ以降のスクリプト内の AutoGen の Call が自動的にインストルメントされます。
```python lines {2} theme={"system"}
import weave
weave.init("autogen-demo")
```
このスニペットを実行すると、Weave は `autogen-demo` project にトレースを送信するよう設定され、以降スクリプト内で発生する AutoGen のアクティビティはすべて Weave によって自動的に取得されます。
モデルクライアントのトレース
以下のセクションでは、AutoGen 内のモデルクライアントに直接行われた Call を Weave がどのようにトレースするかを、単一の Call、ストリーミング応答、キャッシュされた応答のそれぞれについて説明します。
クライアントの create Call のトレース
この例では、`OpenAIChatCompletionClient` への Call をトレースする方法を示します。
```python lines theme={"system"}
import asyncio
from autogen_core.models import UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient
# from autogen_ext.models.anthropic import AnthropicChatCompletionClient
async def simple_client_call(model_name = "gpt-4o"):
model_client = OpenAIChatCompletionClient(
model=model_name,
)
# Anthropic など、他のモデルクライアントを使用することもできます
# model_client = AnthropicChatCompletionClient(
# model="claude-3-haiku-20240307"
# )
response = await model_client.create(
[UserMessage(content="Hello, how are you?", source="user")]
)
print(response)
asyncio.run(simple_client_call())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee09-8dcf-7b72-8cdc-7699608cd6ef%3FhideTraceTree%3D0)
ストリーミングを使用したクライアントの create Call のトレース
Weave は、ストリーミング応答のトレースもサポートしています。
```python lines theme={"system"}
async def simple_client_call_stream(model_name = "gpt-4o"):
openai_model_client = OpenAIChatCompletionClient(model=model_name)
async for item in openai_model_client.create_stream(
[UserMessage(content="Hello, how are you?", source="user")]
):
print(item, flush=True, end="")
asyncio.run(simple_client_call_stream())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee0e-24be-7523-b15e-04f87c03ac68%3FhideTraceTree%3D0)
キャッシュされたクライアント Call のトレース
AutoGen の `ChatCompletionCache` を使用することもできます。Weave はこれらのやり取りをトレースし、応答がキャッシュから返されたのか、新しい Call によるものなのかを表示します。
```python lines theme={"system"}
from autogen_ext.models.cache import ChatCompletionCache
async def run_cache_client(model_name = "gpt-4o"):
openai_model_client = OpenAIChatCompletionClient(model=model_name)
cache_client = ChatCompletionCache(openai_model_client,)
response = await cache_client.create(
[UserMessage(content="Hello, how are you?", source="user")]
)
print(response) # OpenAI からの応答が出力されます
response = await cache_client.create(
[UserMessage(content="Hello, how are you?", source="user")]
)
print(response) # キャッシュされた応答が出力されます
asyncio.run(run_cache_client())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee11-fded-72c2-baaa-7c0ba2a7cd3b%3FhideTraceTree%3D0)
このセクションでは、Weave がエージェントとそのツールの使用をトレースする方法を説明します。これにより、エージェントがどのようにツールを選択して実行するかを把握できます。次の例では、天気ツールを定義し、`AssistantAgent` に割り当てます。
```python lines theme={"system"}
from autogen_agentchat.agents import AssistantAgent
async def get_weather(city: str) -> str:
return f"The weather in {city} is 73 degrees and Sunny."
async def run_agent_with_tools(model_name = "gpt-4o"):
model_client = OpenAIChatCompletionClient(model=model_name)
agent = AssistantAgent(
name="weather_agent",
model_client=model_client,
tools=[get_weather],
system_message="You are a helpful assistant.",
reflect_on_tool_use=True,
)
# 出力をコンソールにストリーミングする場合:
# await Console(agent.run_stream(task="What is the weather in New York?"))
res = await agent.run(task="What is the weather in New York?")
print(res)
await model_client.close()
asyncio.run(run_agent_with_tools())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee13-e5ca-72a1-b7b6-4b263fad89e3%3FhideTraceTree%3D0)
ラウンドロビン方式の GroupChat のトレース
Weave は `RoundRobinGroupChat` などのグループチャット内のやり取りをトレースするため、エージェント間の会話の流れを追跡できます。すべてのエージェントのターンを 1 つの親トレースの下にまとめて効率よく確認するには、グループチャットの関数を `@weave.op` でラップします。この手順は省略可能ですが、実施することをお勧めします。
```python lines theme={"system"}
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
# グループチャット全体をトレースするために、ここで Weave Op を追加しています
# 必須ではありませんが、使用することを強く推奨します
@weave.op
async def run_round_robin_group_chat(model_name="gpt-4o"):
model_client = OpenAIChatCompletionClient(model=model_name)
primary_agent = AssistantAgent(
"primary",
model_client=model_client,
system_message="You are a helpful AI assistant.",
)
critic_agent = AssistantAgent(
"critic",
model_client=model_client,
system_message="Provide constructive feedback. Respond with 'APPROVE' to when your feedbacks are addressed.",
)
text_termination = TextMentionTermination("APPROVE")
team = RoundRobinGroupChat(
[primary_agent, critic_agent], termination_condition=text_termination
)
await team.reset()
# 出力をコンソールにストリーミングする場合:
# await Console(team.run_stream(task="Write a short poem about the fall season."))
result = await team.run(task="Write a short poem about the fall season.")
print(result)
await model_client.close()
asyncio.run(run_round_robin_group_chat())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?filter=%7B%22opVersionRefs%22%3A%5B%22weave%3A%2F%2F%2Fparambharat%2Fautogen-demo%2Fop%2Frun_round_robin_group_chat%3A*%22%5D%7D\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196f16c-26ce-7b32-8f0c-2366d29038a3%3FdescendentCallId%3D0196f16c-26ce-7b32-8f0c-2366d29038a3%26hideTraceTree%3D0)
メモリのトレース
Weave は AutoGen のメモリコンポーネントをトレースできます。`@weave.op()` を使用してメモリ操作を 1 つのトレースにまとめると、可読性が向上します。これにより、メモリの追加や取得の Call が、それらを使用するエージェントの run と一緒に表示されるようになります。
```python lines theme={"system"}
from autogen_core.memory import ListMemory, MemoryContent, MemoryMimeType
# メモリの add Call と get Call を 1 つのトレースにまとめて
# トレースするため、ここに Weave Op を追加しています
# 使用は任意ですが、強く推奨します
@weave.op
async def run_memory_agent(model_name="gpt-4o"):
user_memory = ListMemory()
await user_memory.add(
MemoryContent(
content="The weather should be in metric units",
mime_type=MemoryMimeType.TEXT,
)
)
await user_memory.add(
MemoryContent(
content="Meal recipe must be vegan", mime_type=MemoryMimeType.TEXT
)
)
async def get_weather(city: str, units: str = "imperial") -> str:
if units == "imperial":
return f"The weather in {city} is 73 °F and Sunny."
elif units == "metric":
return f"The weather in {city} is 23 °C and Sunny."
else:
return f"Sorry, I don't know the weather in {city}."
model_client = OpenAIChatCompletionClient(model=model_name)
assistant_agent = AssistantAgent(
name="assistant_agent",
model_client=model_client,
tools=[get_weather],
memory=[user_memory],
)
# 出力をコンソールにストリーミングする場合:
# stream = assistant_agent.run_stream(task="What is the weather in New York?")
# await Console(stream)
result = await assistant_agent.run(task="What is the weather in New York?")
print(result)
await model_client.close()
asyncio.run(run_memory_agent())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee18-28b6-7063-90df-77aaedf88dc9%3FhideTraceTree%3D0)
RAG ワークフローのトレース
Weave では、`ChromaDBVectorMemory` などのメモリシステムを使用したドキュメントのインデックス作成や取得を含め、取得拡張生成 (RAG) ワークフローをトレースできます。フロー全体を可視化し、インデックス作成、取得、そしてその結果として実行される LLM Call を 1 つのトレースにまとめて表示するには、RAG プロセスを `@weave.op()` でデコレートします。
この RAG の例を実行するには `chromadb` が必要です。`pip install chromadb` でインストールしてください。
```python lines {1,65} theme={"system"}
# !pip install -q chromadb
# 環境に chromadb がインストールされていることを確認してください: `pip install chromadb`
import re
from typing import List
import os
from pathlib import Path
import aiofiles
import aiohttp
from autogen_core.memory import Memory, MemoryContent, MemoryMimeType
from autogen_ext.memory.chromadb import (
ChromaDBVectorMemory,
PersistentChromaDBVectorMemoryConfig,
)
class SimpleDocumentIndexer:
def __init__(self, memory: Memory, chunk_size: int = 1500) -> None:
self.memory = memory
self.chunk_size = chunk_size
async def _fetch_content(self, source: str) -> str:
if source.startswith(("http://", "https://")):
async with aiohttp.ClientSession() as session:
async with session.get(source) as response:
return await response.text()
else:
async with aiofiles.open(source, "r", encoding="utf-8") as f:
return await f.read()
def _strip_html(self, text: str) -> str:
text = re.sub(r"<[^>]*>", " ", text)
text = re.sub(r"\\s+", " ", text)
return text.strip()
def _split_text(self, text: str) -> List[str]:
chunks: list[str] = []
for i in range(0, len(text), self.chunk_size):
chunk = text[i : i + self.chunk_size]
chunks.append(chunk.strip())
return chunks
async def index_documents(self, sources: List[str]) -> int:
total_chunks = 0
for source in sources:
try:
content = await self._fetch_content(source)
if "<" in content and ">" in content:
content = self._strip_html(content)
chunks = self._split_text(content)
for i, chunk in enumerate(chunks):
await self.memory.add(
MemoryContent(
content=chunk,
mime_type=MemoryMimeType.TEXT,
metadata={"source": source, "chunk_index": i},
)
)
total_chunks += len(chunks)
except Exception as e:
print(f"Error indexing {source}: {str(e)}")
return total_chunks
@weave.op
async def run_rag_agent(model_name="gpt-4o"):
rag_memory = ChromaDBVectorMemory(
config=PersistentChromaDBVectorMemoryConfig(
collection_name="autogen_docs",
persistence_path=os.path.join(str(Path.home()), ".chromadb_autogen_weave"),
k=3,
score_threshold=0.4,
)
)
# await rag_memory.clear() # 既存のメモリをクリアする場合は、コメントを解除してください
async def index_autogen_docs() -> None:
indexer = SimpleDocumentIndexer(memory=rag_memory)
sources = [
"https://raw.githubusercontent.com/microsoft/autogen/main/README.md",
"https://microsoft.github.io/autogen/dev/user-guide/agentchat-user-guide/tutorial/agents.html",
]
chunks: int = await indexer.index_documents(sources)
print(f"Indexed {chunks} chunks from {len(sources)} AutoGen documents")
# インデックスを作成するのは、コレクションが空の場合か再作成したい場合のみにします
# デモでは、毎回インデックスを作成しても、作成済みかどうかを確認してもかまいません。
# この例では実行のたびにインデックスを作成します。確認処理の追加を検討してください。
await index_autogen_docs()
model_client = OpenAIChatCompletionClient(model=model_name)
rag_assistant = AssistantAgent(
name="rag_assistant",
model_client=model_client,
memory=[rag_memory],
)
# コンソールにストリーミング出力する場合:
# stream = rag_assistant.run_stream(task="What is AgentChat?")
# await Console(stream)
result = await rag_assistant.run(task="What is AgentChat?")
print(result)
await rag_memory.close()
await model_client.close()
asyncio.run(run_rag_agent())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee1b-bac5-7b80-8be7-6e6ea7d1d63d%3FhideTraceTree%3D0)
エージェントランタイムのトレース
Weave は、`SingleThreadedAgentRuntime` などの AutoGen のエージェントランタイム内の操作をトレースできます。ランタイムの実行関数を `@weave.op()` でラップすると関連するトレースがグループ化され、ランタイムの実行中に呼び出されるメッセージハンドラの一連の流れを全体として確認できます。
```python lines {50} theme={"system"}
from dataclasses import dataclass
from typing import Callable
from autogen_core import (
DefaultTopicId,
MessageContext,
RoutedAgent,
default_subscription,
message_handler,
AgentId,
SingleThreadedAgentRuntime
)
@dataclass
class Message:
content: int
@default_subscription
class Modifier(RoutedAgent):
def __init__(self, modify_val: Callable[[int], int]) -> None:
super().__init__("A modifier agent.")
self._modify_val = modify_val
@message_handler
async def handle_message(self, message: Message, ctx: MessageContext) -> None:
val = self._modify_val(message.content)
print(f"{'-'*80}\\nModifier:\\nModified {message.content} to {val}")
await self.publish_message(Message(content=val), DefaultTopicId())
@default_subscription
class Checker(RoutedAgent):
def __init__(self, run_until: Callable[[int], bool]) -> None:
super().__init__("A checker agent.")
self._run_until = run_until
@message_handler
async def handle_message(self, message: Message, ctx: MessageContext) -> None:
if not self._run_until(message.content):
print(f"{'-'*80}\\nChecker:\\n{message.content} passed the check, continue.")
await self.publish_message(
Message(content=message.content), DefaultTopicId()
)
else:
print(f"{'-'*80}\\nChecker:\\n{message.content} failed the check, stopping.")
# エージェントランタイムの呼び出し全体を 1 つのトレースに
# まとめるため、ここで Weave Op を追加しています
# 必須ではありませんが、使用することを強く推奨します
@weave.op
async def run_agent_runtime() -> None:
runtime = SingleThreadedAgentRuntime()
await Modifier.register(
runtime,
"modifier",
lambda: Modifier(modify_val=lambda x: x - 1),
)
await Checker.register(
runtime,
"checker",
lambda: Checker(run_until=lambda x: x <= 1),
)
runtime.start()
await runtime.send_message(Message(content=3), AgentId("checker", "default"))
await runtime.stop_when_idle()
asyncio.run(run_agent_runtime())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee1d-6246-7f11-afb1-3a1874f79023%3FhideTraceTree%3D0)
逐次ワークフローのトレース
エージェント間のやり取りを順に定義する、複雑なエージェントワークフローをトレースできます。`@weave.op()` を使用してワークフロー全体を上位レベルでトレースすると、各エージェントの処理が 1 つの親 Call の下にネストされます。次の例では、コンセプト抽出エージェント、ライターエージェント、書式設定・校正エージェント、ユーザーエージェントをチェーンとしてつなぎ、洗練されたマーケティングコピーを作成します。
```python lines {108} theme={"system"}
from autogen_core import TopicId, type_subscription
from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage
@dataclass
class WorkflowMessage:
content: str
concept_extractor_topic_type = "ConceptExtractorAgent"
writer_topic_type = "WriterAgent"
format_proof_topic_type = "FormatProofAgent"
user_topic_type = "User"
@type_subscription(topic_type=concept_extractor_topic_type)
class ConceptExtractorAgent(RoutedAgent):
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("A concept extractor agent.")
self._system_message = SystemMessage(
content=(
"You are a marketing analyst. Given a product description, identify:\n"
"- Key features\n"
"- Target audience\n"
"- Unique selling points\n\n"
)
)
self._model_client = model_client
@message_handler
async def handle_user_description(self, message: WorkflowMessage, ctx: MessageContext) -> None:
prompt = f"Product description: {message.content}"
llm_result = await self._model_client.create(
messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)],
cancellation_token=ctx.cancellation_token,
)
response = llm_result.content
assert isinstance(response, str)
print(f"{'-'*80}\\n{self.id.type}:\\n{response}")
await self.publish_message(
WorkflowMessage(response), topic_id=TopicId(writer_topic_type, source=self.id.key)
)
@type_subscription(topic_type=writer_topic_type)
class WriterAgent(RoutedAgent):
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("A writer agent.")
self._system_message = SystemMessage(
content=(
"You are a marketing copywriter. Given a block of text describing features, audience, and USPs, "
"compose a compelling marketing copy (like a newsletter section) that highlights these points. "
"Output should be short (around 150 words), output just the copy as a single text block."
)
)
self._model_client = model_client
@message_handler
async def handle_intermediate_text(self, message: WorkflowMessage, ctx: MessageContext) -> None:
prompt = f"Below is the info about the product:\\n\\n{message.content}"
llm_result = await self._model_client.create(
messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)],
cancellation_token=ctx.cancellation_token,
)
response = llm_result.content
assert isinstance(response, str)
print(f"{'-'*80}\\n{self.id.type}:\\n{response}")
await self.publish_message(
WorkflowMessage(response), topic_id=TopicId(format_proof_topic_type, source=self.id.key)
)
@type_subscription(topic_type=format_proof_topic_type)
class FormatProofAgent(RoutedAgent):
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("A format & proof agent.")
self._system_message = SystemMessage(
content=(
"You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, "
"give format and make it polished. Output the final improved copy as a single text block."
)
)
self._model_client = model_client
@message_handler
async def handle_intermediate_text(self, message: WorkflowMessage, ctx: MessageContext) -> None:
prompt = f"Draft copy:\\n{message.content}."
llm_result = await self._model_client.create(
messages=[self._system_message, UserMessage(content=prompt, source=self.id.key)],
cancellation_token=ctx.cancellation_token,
)
response = llm_result.content
assert isinstance(response, str)
print(f"{'-'*80}\\n{self.id.type}:\\n{response}")
await self.publish_message(
WorkflowMessage(response), topic_id=TopicId(user_topic_type, source=self.id.key)
)
@type_subscription(topic_type=user_topic_type)
class UserAgent(RoutedAgent):
def __init__(self) -> None:
super().__init__("A user agent that outputs the final copy to the user.")
@message_handler
async def handle_final_copy(self, message: WorkflowMessage, ctx: MessageContext) -> None:
print(f"\\n{'-'*80}\\n{self.id.type} received final copy:\\n{message.content}")
# エージェントのワークフロー全体を 1 つのトレースにまとめて
# トレースするため、ここに Weave Op を追加しています
# 使用は任意ですが、使用することを強く推奨します
@weave.op(call_display_name="Sequential Agent Workflow")
async def run_agent_workflow(model_name="gpt-4o"):
model_client = OpenAIChatCompletionClient(model=model_name)
runtime = SingleThreadedAgentRuntime()
await ConceptExtractorAgent.register(runtime, type=concept_extractor_topic_type, factory=lambda: ConceptExtractorAgent(model_client=model_client))
await WriterAgent.register(runtime, type=writer_topic_type, factory=lambda: WriterAgent(model_client=model_client))
await FormatProofAgent.register(runtime, type=format_proof_topic_type, factory=lambda: FormatProofAgent(model_client=model_client))
await UserAgent.register(runtime, type=user_topic_type, factory=lambda: UserAgent())
runtime.start()
await runtime.publish_message(
WorkflowMessage(
content="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours"
),
topic_id=TopicId(concept_extractor_topic_type, source="default"),
)
await runtime.stop_when_idle()
await model_client.close()
asyncio.run(run_agent_workflow())
```
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196ee1f-dd53-73f2-9119-2a44da92c5ae%3FhideTraceTree%3D0)
コード実行器のトレース
**Docker が必須です**
この例では Docker を使用してコードを実行するため、環境によっては動作しない場合があります (たとえば、Colab 上で直接実行する場合) 。試す際は、ローカルで Docker が起動していることを確認してください。
Weave は、AutoGen エージェントによるコードの生成と実行をトレースします。アシスタントエージェントが生成したコードと、そのコードを実行した際に実行器エージェントが返す出力の両方を確認できます。
````python lines {69} theme={"system"}
import tempfile
from autogen_core import DefaultTopicId
from autogen_core.code_executor import CodeBlock, CodeExecutor
from autogen_core.models import (
AssistantMessage,
ChatCompletionClient,
LLMMessage,
SystemMessage,
UserMessage,
)
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
@dataclass
class CodeGenMessage:
content: str
@default_subscription
class Assistant(RoutedAgent):
def __init__(self, model_client: ChatCompletionClient) -> None:
super().__init__("An assistant agent.")
self._model_client = model_client
self._chat_history: List[LLMMessage] = [
SystemMessage(
content="""Write Python script in markdown block, and it will be executed.
Always save figures to file in the current directory. Do not use plt.show(). All code required to complete this task must be contained within a single response.""",
)
]
@message_handler
async def handle_message(self, message: CodeGenMessage, ctx: MessageContext) -> None:
self._chat_history.append(UserMessage(content=message.content, source="user"))
result = await self._model_client.create(self._chat_history)
print(f"\\n{'-'*80}\\nAssistant:\\n{result.content}")
self._chat_history.append(AssistantMessage(content=result.content, source="assistant"))
await self.publish_message(CodeGenMessage(content=result.content), DefaultTopicId())
def extract_markdown_code_blocks(markdown_text: str) -> List[CodeBlock]:
pattern = re.compile(r"```(?:\\s*([\\w\\+\\-]+))?\\n([\\s\\S]*?)```")
matches = pattern.findall(markdown_text)
code_blocks: List[CodeBlock] = []
for match in matches:
language = match[0].strip() if match[0] else ""
code_content = match[1]
code_blocks.append(CodeBlock(code=code_content, language=language))
return code_blocks
@default_subscription
class Executor(RoutedAgent):
def __init__(self, code_executor: CodeExecutor) -> None:
super().__init__("An executor agent.")
self._code_executor = code_executor
@message_handler
async def handle_message(self, message: CodeGenMessage, ctx: MessageContext) -> None:
code_blocks = extract_markdown_code_blocks(message.content)
if code_blocks:
result = await self._code_executor.execute_code_blocks(
code_blocks, cancellation_token=ctx.cancellation_token
)
print(f"\\n{'-'*80}\\nExecutor:\\n{result.output}")
await self.publish_message(CodeGenMessage(content=result.output), DefaultTopicId())
# コード生成ワークフロー全体を 1 つのトレースとして記録するため、
# ここに Weave Op を追加しています
# 必須ではありませんが、使用を強く推奨します
@weave.op(call_display_name="CodeGen Agent Workflow")
async def run_codegen(model_name="gpt-4o"): # モデルを更新
work_dir = tempfile.mkdtemp()
runtime = SingleThreadedAgentRuntime()
# この例を実行するには、Docker が起動していることを確認してください
try:
async with DockerCommandLineCodeExecutor(work_dir=work_dir) as executor:
model_client = OpenAIChatCompletionClient(model=model_name)
await Assistant.register(runtime, "assistant", lambda: Assistant(model_client=model_client))
await Executor.register(runtime, "executor", lambda: Executor(executor))
runtime.start()
await runtime.publish_message(
CodeGenMessage(content="Create a plot of NVDA vs TSLA stock returns YTD from 2024-01-01."),
DefaultTopicId(),
)
await runtime.stop_when_idle()
await model_client.close()
except Exception as e:
print(f"Could not run Docker code executor example: {e}")
print("Please ensure Docker is installed and running.")
finally:
import shutil
shutil.rmtree(work_dir)
asyncio.run(run_codegen())
````
[
](https://forge.coreweave.com/wandb/parambharat/autogen-demo/weave/traces?view=traces_default\&peekPath=%2Fparambharat%2Fautogen-demo%2Fcalls%2F0196f173-21c2-7540-9dc7-fbab0b94ce0e%3FhideTraceTree%3D0)
詳細情報
Weave のトレースと Op について理解を深めるには、以下のガイドを参照してください。
* **Weave**:
* [トレースガイド](/ja/products/wandb/weave/guides/tracking/tracing)
* [Op デコレーター](/ja/products/wandb/weave/guides/tracking/ops)
* **AutoGen**:
* [公式ドキュメント](https://microsoft.github.io/autogen/stable//index.html)
* [AutoGen GitHub](https://github.com/microsoft/autogen)
このガイドでは、Weave と AutoGen の統合の基本を紹介しました。Weights & Biases の UI では、エージェントのやり取り、モデルの Call、ツールの使用に関する詳細なトレースを確認できます。