AutoGen 初识
autogen 介绍
AutoGen 是一个用于构建多 Agent AI 应用的综合框架，支持 Agent 自主工作或与人类在复杂工作流中协作。AutoGen 作为一个多 Agent 编排框架，为通过协调 Agent 交互构建复杂的 AI 驱动应用提供了基础。该框架支持自主 Agent 操作和人在环场景，使其适用于从自动任务完成到协作问题解决的各种用例。
autogen 既可以作为自主规划型的智能体，也可以使用 GraphFlow 编排复杂的工作流智能体
本系列文档会根据官方的文档，加上我自己的理解，由浅入深的介绍 autogen 的使用
注意
autogen 是微软推出的项目，秉承着微软一贯的作风--烂尾，目前更新的频率很慢，官方最新版本还停留在 2025-09-30 号，已经好几个月没有更新了，最后的 commit 记录显示是 2025-10-4 号，相比于 langchain 和 langgraph，autogen 看着就像是个快要死掉的项目了，我有点担心后面这个项目又会烂尾了。
不过无所谓，我觉得与 langgraph 相比，autogen 会简单一些，我们先来学习一下它的基础使用与核心思想。
先初始化项目
uv init autogen_test --python 3.12
cd autogen_test
uv sync
source .venv/bin/activate
# 安装依赖
uv add autogen-agentchat "autogen-ext[openai]"
定义工具函数
创建 agent_tools.py ，这里创建一个工具，调用高德的天气查询接口，返回某个城市的天气信息。
import datetime
import httpx
import os
from pydantic import BaseModel
from typing import List, Dict, Any
class WeatherInfo(BaseModel):
"""天气信息"""
status: str = "success"
message: str = ""
data: List[Dict[str, Any]] = []
async def get_weather(city: str)-> dict:
"""获取某地天气
Args:
city (str): 要查询天气的城市名称, 例如：北京.
Returns:
dict: 该城市的天气信息 或者 错误信息.
"""
weather_data = WeatherInfo()
api_key = os.getenv("GAODE_KEY", "")
if not api_key:
weather_data.status = "error"
weather_data.message = "未配置高德地图API Key"
return weather_data.model_dump()
api_domain = 'https://restapi.amap.com/v3'
url = f"{api_domain}/config/district?keywords={city}"f"&subdistrict=0&extensions=base&key={api_key}"
headers = {"Content-Type": "application/json; charset=utf-8"}
async with httpx.AsyncClient(headers=headers, verify=False) as client:
response = await client.get(url)
if response.status_code != 200:
weather_data.status = "error"
weather_data.message = "查询失败"
return weather_data.model_dump()
city_info = response.json()
if city_info["info"] != "OK":
weather_data.status = "error"
weather_data.message = "获取城市信息查询失败"
return weather_data.model_dump()
CityCode = city_info['districts'][0]['adcode']
weather_url = f"{api_domain}/weather/weatherInfo?city={CityCode}&extensions=all&key={api_key}"
weatherInfo_response = await client.get(weather_url)
if weatherInfo_response.status_code != 200:
weather_data.message = "查询天气信息失败"
weather_data.status = "error"
return weather_data.model_dump()
weatherInfo = weatherInfo_response.json()
if weatherInfo['info'] != "OK":
weather_data.message = "查询天气信息失败"
weather_data.status = "error"
return weather_data.model_dump()
weatherInfo_data = weatherInfo_response.json()
contents = []
if len(weatherInfo_data.get('forecasts')) <= 0:
weather_data.message = "没有获取到该城市的天气信息"
weather_data.status = "error"
return weather_data.model_dump()
for item in weatherInfo_data['forecasts'][0]['casts']:
content = {
'date': item.get('date'),
'week': item.get('week'),
'dayweather': item.get('dayweather'),
'daytemp_float': item.get('daytemp_float'),
'daywind': item.get('daywind'),
'nightweather': item.get('nightweather'),
'nighttemp_float': item.get('nighttemp_float')
}
contents.append(content)
weather_data.data = contents
weather_data.status = "success"
weather_data.message = "获取天气成功"
return weather_data.model_dump()
编辑 main.py
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelFamily, ModelInfo
from agent_tools import get_weather
model_info = ModelInfo(
vision=False,
function_calling=True,
json_output=True,
structured_output=True,
family=ModelFamily.UNKNOWN,
multiple_system_messages=True
)
model_client = OpenAIChatCompletionClient(
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model_info=model_info
)
# 定义一个助手代理
agent = AssistantAgent(
name="weather_agent",
model_client=model_client,
tools=[get_weather],
system_message="你是一个助理，回答问题时请使用中文作答",
reflect_on_tool_use=True,
model_client_stream=True
async def main() -> None:
await Console(agent.run_stream(task="北京明天的天气"))
await model_client.close()
if __name__ == "__main__":
asyncio.run(main())
这里我使用的是 qwen-max 模型，从我测试的结果来看，这个模型效果相对较好，创建环境变量文件 .env，写入程序中需要的各种 key
OPENAI_API_KEY=xxxx
GAODE_KEY=xxxx
运行
uv run --env-file=.env main.py
得到以下输出
---------- TextMessage (user) ----------
北京明天的天气怎么样?
---------- ToolCallRequestEvent (weather_agent) ----------
[FunctionCall(id='call_e4ce6604ee544338be8e8f', arguments='{"city": "北京"}', name='get_weather')]
---------- ToolCallExecutionEvent (weather_agent) ----------
[FunctionExecutionResult(content="{'status': 'success', 'message': '获取天气成功', 'data': [{'date': '2026-01-05', 'week': '1', 'dayweather': '晴', 'daytemp_float': '6.0', 'daywind': '西南', 'nightweather': ' 晴', 'nighttemp_float': '-5.0'}, {'date': '2026-01-06', 'week': '2', 'dayweather': '多云', 'daytemp_float': '7.0', 'daywind': '西北', 'nightweather': '多云', 'nighttemp_float': '-5.0'}, {'date': '2026-01-07', 'week': '3', 'dayweather': '晴', 'daytemp_float': '4.0', 'daywind': '北', 'nightweather': '晴', 'nighttemp_float': '-6.0'}, {'date': '2026-01-08', 'week': '4', 'dayweather': '晴', 'daytemp_float': '5.0', 'daywind': '南', 'nightweather': '多云', 'nighttemp_float': '-3.0'}]}", name='get_weather', call_id='call_e4ce6604ee544338be8e8f', is_error=False)]
---------- ModelClientStreamingChunkEvent (weather_agent) ----------
北京明天（2026年1月6日）的天气预报是多云，白天最高气温约为7.0℃，夜间最低气温约为-5.0℃。风向为西北风。请根据天气变化适当调整着装。
从结果来看程序的运行
- agent 先接收用户输入的问题 TextMessage (user) 北京明天的天气怎么样?
- 模型根据 agent 的 tool 信息，触发 ToolCallRequestEvent， 返回 fc 调用的参数
- agent 调用 get_weather 工具， 触发 ToolCallExecutionEvent 事件，返回函数的结果
- agent 再次调用大模型进行总结，触发 ModelClientStreamingChunkEvent 事件，返回最终的输出。
这个 agent 是一个最简单的 agent，只有一个工具，没有工作流，只能干查询天气的事，但是在实际使用过程中，还是遇到很多问题的。
遇到的问题
调用 autogen 不知道的模型
autogen 有一个模型的列表在
autogen_ext\models\openai\_model_info.py
_MODEL_INFO: Dict[str, ModelInfo] = {
...
"gpt-4o-mini": {
"vision": True,
"function_calling": True,
"json_output": True,
"family": ModelFamily.GPT_4O,
"structured_output": True,
"multiple_system_messages": True,
},
...
}
这个字典是模型和这个模型的能力(ModelInfo) 的关系，比如vision 表示这个模型是否具备图像识别能力，autogen 会根据模型的能力先做一些判断，如果用户尝试使用图像识别，但是模型没有vision能力，那么就会抛异常，如果使用一个不在列表中的模型，如上面代码中使用的 qwen-max, 这时就需要添加上 model_info 参数，将模型能力告诉 autogen。
model_info = ModelInfo(
vision=False,
function_calling=True,
json_output=True,
structured_output=True,
family=ModelFamily.UNKNOWN,
multiple_system_messages=True
)
model_client = OpenAIChatCompletionClient(
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model_info=model_info
)
model_info 为描述某个模型能力的说明类
class ModelInfo(TypedDict, total=False):
vision: Required[bool]
function_calling: Required[bool]
json_output: Required[bool]
family: Required[ModelFamily.ANY | str]
structured_output: Required[bool]
multiple_system_messages: Optional[bool]
当用到了一个autogen 不认识的模型时，需要告诉autogen 这个模型有什么能力
- vision: 是否有视觉能力，如图像识别
- function_calling: 是否支持 function_calling
- json_output: 是否支持json 输出
- structured_output: 是否支持结构化输出
- family: 模型家族，这指的是一类模型，比如gpt-4o-mini-search-preview-2025-03-11和gpt-4o-search-preview-2025-03-11 都是 gpt-4o 家族，它们能力上是相同的，只是版本不同，不知道的可以设置为ModelFamily.UNKNOWN，也可以自定义一个，只是 autogen 底层当UNKNOWN 来处理
- multiple_system_messages: 模型支持多个非连续的系统消息，这个是可选的，openai 系列的模型大多数的支持的，Claude 模型是 不支持 的
结构化输出和json输出的区别
ModelInfo 中有个json_output(JSON 输出)和structured_output(结构化输出)， 这两个能力其实有一定的重叠，但它们侧重点不同，适用场景也有差异。下面是两者的对比说明：
| 特性 | 结构化输出 | JSON 输出 |
|---|---|---|
| 概念 | 一种规范的输出格式，可以是JSON，也可以是表格、列表、XML等。强调结构清晰，便于机器解析。 | 一种具体的数据格式，是结构化输出的一种形式。使用键值对表达复杂数据。 |
| 输出格式 | 可以是JSON、Markdown表格、HTML等结构化形式。 | 固定为JSON格式，如：{"name": "Tom", "age": 20} |
| 目的 | 让模型输出特定的结构，便于后处理或程序消费。 | 更偏向于程序可读性和标准化数据交换（尤其在API中广泛使用）。 |
| 控制方式 | 通过prompt模板、函数调用（function calling）、tools 等方式指导模型生成结构化内容。 | 通常是结构化输出的一种，但也可以用函数调用或 prompt 模板指定为JSON格式。 |
模型供应商
上面使用 autogen-ext[openai] 在安装autogen-ext 的同时安装了openai 这个供应商需要的依赖，autogen-ext 默认支持 openai, azure,anthropic, ollama, llama_cpp, replay 等模型供应商，基本上常见的供应商都支持了，可以用到哪个装哪个，具体可以查看 https://github.com/microsoft/autogen/blob/main/python/packages/autogen-ext/pyproject.toml 这个文件中支持的模型供应商。
有个好玩的现象，google 家的 adk 默认不支持 openai 的，这个autogen 默认也不支持 gemini 的， 大厂直接的隔阂真大呀
工具说明与数据的返回
工具的说明采用函数注释的方式，需要将工具的名称，描述，入参，返回值写清晰
async def get_weather(city: str)-> dict:
"""获取某地天气
Args:
city (str): 要查询天气的城市名称, 例如：北京.
Returns:
dict: 该城市的天气信息 或者 错误信息.
"""
返回值可以是字符串或者进行序列化的数据，我这里使用 pydantic 中的BaseModel
class WeatherInfo(BaseModel):
"""天气信息"""
status: str = "success"
message: str = ""
data: List[Dict[str, Any]] = []
这里最好设置一个用于表现数据状态的字段 status, 且清晰的用文字表达， 如 success, error, fail 等，而不是业务代码如 0， 1， 3 这种，数字状态码交给大模型也不会理解的。
如何更好的定义工具？
上面的天气查询工具很简单，就是查询某地天气，参数只有一个，所以将这些内容写到函数的注释中即可，对于复杂的工具函数，如参数非必填，参数有默认值，参数的取值需要在某个范围之内，应该如何定义工具函数？
先来看看 agent 中 tools 的签名
def __init__(
self,
name: str,
model_client: ChatCompletionClient,
*,
tools: List[BaseTool[Any, Any] | Callable[..., Any] | Callable[..., Awaitable[Any]]] | None = None,
workbench: Workbench | Sequence[Workbench] | None = None,
...
tools 参数可以是 BaseTool 类，Callable 或者 Awaitable 的列表，Callable 可以简单理解为普通的 python 函数(def )或者 python 类(这个类要实现 def __call__ 方法)， Awaitable 是协程函数(async def) 或者实现了 async def __acall__ 方法的类。
使用 def 或者 async def 定义的工具函数是最普遍的方法，这个方法需要将工具能力说明和参数说明写到注释中
for tool in tools:
if isinstance(tool, BaseTool):
self._tools.append(tool)
elif callable(tool):
if hasattr(tool, "__doc__") and tool.__doc__ is not None:
description = tool.__doc__
else:
description = ""
self._tools.append(FunctionTool(tool, description=description))
else:
raise ValueError(f"Unsupported tool type: {type(tool)}")
查看源码，最终是会使用 FunctionTool 来包一层的。
但我始终认为将工具说明和参数写到注释中并不是一个很好的方式，我更希望明确的工具函数说明，当然如果查看 autogen 源代码，也是会将函数的参数转换为 BaseModel 的，来简化一下，查看 FunctionTool 到底是怎么解析参数的。
FunctionTool 类解析工具参数
from autogen_core.tools import FunctionTool
async def get_weather(city: str, day: int = 1) -> dict:
"""获取某地天气
Args:
city (str): 要查询天气的城市名称, 例如：北京.
day (int): 要查询哪天的，0为明天，1为所有，默认为 1
Returns:
dict: 该城市的天气信息 或者 错误信息.
:param day:
"""
pass
ft = FunctionTool(get_weather, description=get_weather.__doc__)
args_schema = ft.args_type().model_json_schema()
print(json.dumps(args_schema, indent=4, ensure_ascii=False))
得到的输出如下
{
"properties": {
"city": {
"description": "city",
"title": "City",
"type": "string"
},
"day": {
"default": 1,
"description": "day",
"title": "Day",
"type": "integer"
}
},
"required": [
"city"
],
"title": "get_weatherargs",
"type": "object"
}
可以看到，通过 FunctionTool 来自动解析函数得到的 args_type 并没有包含函数参数的含义，只是原样的输出的函数参数的名字，如果参数名称与实际含义匹配性不高，那么很有可能影响大模型解析。
接下来我们看一下如何使用 BaseTool 来自定义工具
使用 BaseTool 来自定义工具
BaseTool 是一个抽象基类，定义了工具的基本接口，实现类必须实现的抽象方法即可
async def run(self, args: ArgsT, cancellation_token: CancellationToken) -> ReturnT
我们将之前的 get_weather 工具函数改造成 BaseTool 的继承类
from typing import List, Dict, Any
from autogen_core.tools import BaseTool
from autogen_core import CancellationToken
from pydantic import BaseModel, Field
class WeatherInfo(BaseModel):
"""天气信息"""
status: str = "success"
message: str = ""
data: List[Dict[str, Any]] = []
class WeatherInput(BaseModel):
"""天气查询参数"""
city: str = Field(..., description="要查询天气的城市名称, 例如：北京.")
day: int = Field(1, description="要查询哪天的，0为明天，1为所有，默认为 1")
class GaodeWeather(BaseTool[WeatherInput, WeatherInfo]):
def __init__(self):
super().__init__(
name="get_weather",
description="获取某地天气",
args_type=WeatherInput,
return_type=WeatherInfo
)
async def run(self, args: WeatherInput, cancellation_token: CancellationToken) -> WeatherInfo:
weather_data = WeatherInfo()
api_key = os.getenv("GAODE_KEY", "")
...
先使用BaseModel 定义工具函数的输入和输出，这里可以详细的描述一下参数的含义 ,也可以设置默认值，如果不设置默认值则改参数是必填项
city: str = Field(..., description="要查询天气的城市名称, 例如：北京.")
day: int = Field(1, description="要查询哪天的，0为明天，1为所有，默认为 1")
之后定义 GaodeWeather 工具类
class GaodeWeather(BaseTool[WeatherInput, WeatherInfo]):
def __init__(self):
super().__init__(
name="get_weather",
description="获取某地天气",
args_type=WeatherInput,
return_type=WeatherInfo
)
使用 super().__init__ 方法，将参数类型与工具返回值类型传递给 BaseTool 类，这里还要传 name 和 description 参数。
- name 和 description 用于大模型来判断是否需要调用该工具
- args_type 用于大模型解析参数
我们来看一下使用 BaseTool 以后的 schema 信息
g = GaodeWeather()
args_schema = g.args_type().model_json_schema()
print(json.dumps(args_schema, indent=4, ensure_ascii=False))
输出为
{
"description": "天气查询参数",
"properties": {
"city": {
"description": "要查询天气的城市名称, 例如：北京.",
"title": "City",
"type": "string"
},
"day": {
"default": 1,
"description": "要查询哪天的，0为明天，1为所有，默认为 1",
"title": "Day",
"type": "integer"
}
},
"required": [
"city"
],
"title": "WeatherInput",
"type": "object"
}
此时函数的描述信息就可以很好的被提取出来了。
在 agent 中使用 BaseTool 类
agent = AssistantAgent(
name="weather_agent",
model_client=model_client,
tools=[GaodeWeather()],
system_message="你是一个助理，回答问题时请使用中文作答",
reflect_on_tool_use=True,
model_client_stream=True
)
这里注意，不能像工具函数那样只传函数名，这里要传实例，GaodeWeather()
FunctionTool 和 BaseTool 都可以实现工具，它们适用的范围不太一样
- 继承
BaseTool适合复杂逻辑、需要状态管理或自定义配置的工具 FunctionTool更适合简单的函数包装
总结
本文根据官方文档，搭载天气查询工具，实现了 autogen 中最简单的 agent，并且详细的介绍了如何使用 BaseTool 自定义工具，下一篇准备讲一下智能体 Teams 的使用，多个智能体组成一个团队来完成相应的工作。